<?php

declare(strict_types=1);

namespace CrocoAI\DataBridge\Infrastructure;

use RuntimeException;

final class Repository
{
    private \wpdb $db;

    public function __construct()
    {
        global $wpdb;
        $this->db = $wpdb;
    }

    public function sources(): array
    {
        $rows = $this->db->get_results($this->db->prepare(
            'SELECT * FROM %i ORDER BY id DESC',
            Database::table('sources')
        ), ARRAY_A);
        return array_map([$this, 'hydrateSource'], $rows ?: []);
    }

    public function source(int $id): array
    {
        $row = $this->db->get_row($this->db->prepare(
            'SELECT * FROM %i WHERE id=%d',
            Database::table('sources'),
            $id
        ), ARRAY_A);
        if (!$row) {
            throw new RuntimeException('Source not found.');
        }
        return $this->hydrateSource($row);
    }

    public function saveSource(array $data, ?int $id = null): int
    {
        $now = current_time('mysql', true);
        $row = [
            'name' => sanitize_text_field((string) ($data['name'] ?? 'Untitled source')),
            'type' => in_array($data['type'] ?? '', ['csv', 'xml'], true) ? $data['type'] : 'csv',
            'location' => esc_url_raw((string) ($data['location'] ?? '')),
            'config' => Crypto::encrypt(is_array($data['config'] ?? null) ? $data['config'] : []),
            'profile_id' => !empty($data['profile_id']) ? (int) $data['profile_id'] : null,
            'schedule_seconds' => !empty($data['schedule_seconds']) ? max(300, (int) $data['schedule_seconds']) : null,
            'enabled' => empty($data['enabled']) ? 0 : 1,
            'updated_at' => $now,
        ];
        if ($id) {
            $this->db->update(Database::table('sources'), $row, ['id' => $id]);
            return $id;
        }
        $row['created_at'] = $now;
        $this->db->insert(Database::table('sources'), $row);
        return (int) $this->db->insert_id;
    }

    public function saveProfile(array $data, ?int $id = null): int
    {
        $now = current_time('mysql', true);
        $row = [
            'name' => sanitize_text_field((string) ($data['name'] ?? 'Default mapping')),
            'mapping' => wp_json_encode($data['mapping'] ?? []),
            'ownership' => wp_json_encode($data['ownership'] ?? []),
            'updated_at' => $now,
        ];
        if ($id) {
            $current = $this->profile($id);
            $row['version'] = ((int) $current['version']) + 1;
            $this->db->update(Database::table('profiles'), $row, ['id' => $id]);
            return $id;
        }
        $row['version'] = 1;
        $row['created_at'] = $now;
        $this->db->insert(Database::table('profiles'), $row);
        return (int) $this->db->insert_id;
    }

    public function profiles(): array
    {
        $rows = $this->db->get_results($this->db->prepare(
            'SELECT * FROM %i ORDER BY id DESC',
            Database::table('profiles')
        ), ARRAY_A) ?: [];
        return array_map([$this, 'hydrateProfile'], $rows);
    }

    public function profile(int $id): array
    {
        $row = $this->db->get_row($this->db->prepare(
            'SELECT * FROM %i WHERE id=%d',
            Database::table('profiles'),
            $id
        ), ARRAY_A);
        if (!$row) {
            throw new RuntimeException('Mapping profile not found.');
        }
        return $this->hydrateProfile($row);
    }

    public function createRun(int $sourceId, bool $fullFeed = true): int
    {
        $this->db->insert(Database::table('runs'), [
            'source_id' => $sourceId,
            'status' => 'queued',
            'stage' => 'fetch',
            'full_feed' => $fullFeed ? 1 : 0,
            'stats' => '{}',
            'started_at' => current_time('mysql', true),
        ]);
        return (int) $this->db->insert_id;
    }

    public function run(int $id): array
    {
        $row = $this->db->get_row($this->db->prepare(
            'SELECT * FROM %i WHERE id=%d',
            Database::table('runs'),
            $id
        ), ARRAY_A);
        if (!$row) {
            throw new RuntimeException('Import run not found.');
        }
        $row['stats'] = json_decode((string) $row['stats'], true) ?: [];
        return $row;
    }

    public function runs(int $limit = 50): array
    {
        $sql = $this->db->prepare(
            'SELECT * FROM %i ORDER BY id DESC LIMIT %d',
            Database::table('runs'),
            $limit
        );
        return $this->db->get_results($sql, ARRAY_A) ?: [];
    }

    public function updateRun(int $id, array $data): void
    {
        if (isset($data['stats']) && is_array($data['stats'])) {
            $data['stats'] = wp_json_encode($data['stats']);
        }
        $this->db->update(Database::table('runs'), $data, ['id' => $id]);
    }

    public function addRecord(int $runId, int $sourceId, string $externalId, array $raw, array $canonical, string $status, array $errors = []): int
    {
        $this->db->insert(Database::table('records'), [
            'run_id' => $runId,
            'source_id' => $sourceId,
            'external_id' => $externalId,
            'raw' => wp_json_encode($raw, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES),
            'canonical' => wp_json_encode($canonical, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES),
            'content_hash' => hash('sha256', wp_json_encode($canonical)),
            'status' => $status,
            'error' => $errors ? wp_json_encode($errors) : null,
            'created_at' => current_time('mysql', true),
        ]);
        return (int) $this->db->insert_id;
    }

    public function record(int $id): array
    {
        $row = $this->db->get_row($this->db->prepare(
            'SELECT * FROM %i WHERE id=%d',
            Database::table('records'),
            $id
        ), ARRAY_A);
        if (!$row) {
            throw new RuntimeException('Staged record not found.');
        }
        foreach (['raw', 'canonical', 'error'] as $key) {
            $row[$key] = json_decode((string) ($row[$key] ?? ''), true) ?: [];
        }
        return $row;
    }

    public function link(int $sourceId, string $externalId): ?array
    {
        $row = $this->db->get_row($this->db->prepare(
            'SELECT * FROM %i WHERE source_id=%d AND external_id=%s',
            Database::table('links'),
            $sourceId,
            $externalId
        ), ARRAY_A);
        return $row ?: null;
    }

    public function deleteLink(int $sourceId, string $externalId): void
    {
        $this->db->delete(Database::table('links'), [
            'source_id' => $sourceId,
            'external_id' => $externalId,
        ]);
    }

    public function touchLink(int $sourceId, string $externalId, int $productId, string $hash, int $runId, array $snapshot = []): void
    {
        $existing = $this->link($sourceId, $externalId);
        $row = [
            'product_id' => $productId,
            'canonical_hash' => $hash,
            'last_seen_run' => $runId,
            'missing_count' => 0,
            'managed_snapshot' => wp_json_encode($snapshot),
            'updated_at' => current_time('mysql', true),
        ];
        if ($existing) {
            $this->db->update(Database::table('links'), $row, ['id' => (int) $existing['id']]);
        } else {
            $row += ['source_id' => $sourceId, 'external_id' => $externalId, 'created_at' => current_time('mysql', true)];
            $this->db->insert(Database::table('links'), $row);
        }
    }

    public function addChange(
        int $runId,
        ?int $recordId,
        ?int $productId,
        string $action,
        array $diff,
        string $status = 'pending',
        int $priority = 0
    ): int {
        $this->db->insert(Database::table('changes'), [
            'run_id' => $runId,
            'record_id' => $recordId,
            'product_id' => $productId,
            'action' => $action,
            'status' => $status,
            'priority' => max(0, min(3, $priority)),
            'diff' => wp_json_encode($diff, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES),
            'created_at' => current_time('mysql', true),
        ]);
        return (int) $this->db->insert_id;
    }

    public function changes(array $filters = []): array
    {
        $where = '1=1';
        $args = [Database::table('changes')];
        if (!empty($filters['run_id'])) {
            $where .= ' AND run_id=%d';
            $args[] = (int) $filters['run_id'];
        }
        if (!empty($filters['status'])) {
            $where .= ' AND status=%s';
            $args[] = sanitize_key((string) $filters['status']);
        }
        $sql = $this->db->prepare("SELECT * FROM %i WHERE {$where} ORDER BY id DESC LIMIT 500", ...$args);
        $rows = $this->db->get_results($sql, ARRAY_A) ?: [];
        foreach ($rows as &$row) {
            $row['diff'] = json_decode((string) $row['diff'], true) ?: [];
            $row['previous_data'] = json_decode((string) ($row['previous_data'] ?? ''), true) ?: [];
        }
        return $rows;
    }

    public function change(int $id): array
    {
        $rows = $this->db->get_results($this->db->prepare(
            'SELECT * FROM %i WHERE id=%d',
            Database::table('changes'),
            $id
        ), ARRAY_A);
        if (!$rows) {
            throw new RuntimeException('Change not found.');
        }
        $row = $rows[0];
        $row['diff'] = json_decode((string) $row['diff'], true) ?: [];
        $row['previous_data'] = json_decode((string) ($row['previous_data'] ?? ''), true) ?: [];
        return $row;
    }

    public function updateChange(int $id, array $data): void
    {
        foreach (['diff', 'previous_data'] as $key) {
            if (isset($data[$key]) && is_array($data[$key])) {
                $data[$key] = wp_json_encode($data[$key]);
            }
        }
        $this->db->update(Database::table('changes'), $data, ['id' => $id]);
    }

    public function approveRun(int $runId, int $userId): int
    {
        return (int) $this->db->query($this->db->prepare(
            "UPDATE %i SET status='approved', approved_by=%d, approved_at=%s WHERE run_id=%d AND status='pending'",
            Database::table('changes'),
            $userId,
            current_time('mysql', true),
            $runId
        ));
    }

    /**
     * @param array<int,int> $ids
     * @param array<string,mixed> $mediaGate
     */
    public function createPublicationJob(?int $runId, array $ids, int $userId, array $mediaGate = []): int
    {
        $ids = array_values(array_unique(array_filter(array_map('intval', $ids))));
        if ($runId) {
            $active = $this->db->get_var($this->db->prepare(
                "SELECT id FROM %i WHERE run_id=%d AND status IN ('queued','processing','processing_media','paused') LIMIT 1",
                Database::table('jobs'),
                $runId
            ));
            if ($active) {
                throw new RuntimeException(sprintf('Publication job #%d is already active for this run.', (int) $active));
            }
            $total = (int) $this->db->get_var($this->db->prepare(
                "SELECT COUNT(*) FROM %i WHERE run_id=%d AND status IN ('approved','auto_approved')",
                Database::table('changes'),
                $runId
            ));
            $payload = ['mode' => 'run'];
        } else {
            if (!$ids) {
                throw new RuntimeException('At least one approved change is required.');
            }
            $placeholders = implode(',', array_fill(0, count($ids), '%d'));
            $total = (int) $this->db->get_var($this->db->prepare(
                "SELECT COUNT(*) FROM %i WHERE id IN ({$placeholders}) AND status IN ('approved','auto_approved')",
                Database::table('changes'),
                ...$ids
            ));
            $payload = ['mode' => 'selected', 'ids' => $ids];
        }
        if ($mediaGate) {
            $payload['media_gate'] = $mediaGate;
        }
        if ($total < 1) {
            throw new RuntimeException('No approved changes are available for publication.');
        }
        $now = current_time('mysql', true);
        $this->db->insert(Database::table('jobs'), [
            'job_type' => 'publication',
            'run_id' => $runId,
            'status' => 'queued',
            'stage' => 'products',
            'total' => $total,
            'payload' => wp_json_encode($payload),
            'created_by' => $userId,
            'created_at' => $now,
            'updated_at' => $now,
        ]);
        return (int) $this->db->insert_id;
    }

    public function job(int $id): array
    {
        $row = $this->db->get_row($this->db->prepare(
            'SELECT * FROM %i WHERE id=%d',
            Database::table('jobs'),
            $id
        ), ARRAY_A);
        if (!$row) {
            throw new RuntimeException('Publication job not found.');
        }
        $row['payload'] = json_decode((string) ($row['payload'] ?? ''), true) ?: [];
        foreach (['id', 'run_id', 'total', 'processed', 'succeeded', 'failed', 'media_total', 'media_processed', 'media_failed'] as $key) {
            $row[$key] = $row[$key] === null ? null : (int) $row[$key];
        }
        return $row;
    }

    public function jobs(int $limit = 50): array
    {
        $rows = $this->db->get_results($this->db->prepare(
            'SELECT * FROM %i ORDER BY id DESC LIMIT %d',
            Database::table('jobs'),
            max(1, min(100, $limit))
        ), ARRAY_A) ?: [];
        return array_map(function (array $row): array {
            $row['payload'] = json_decode((string) ($row['payload'] ?? ''), true) ?: [];
            return $row;
        }, $rows);
    }

    public function activeJobs(): array
    {
        $rows = $this->db->get_results($this->db->prepare(
            "SELECT * FROM %i WHERE status IN ('queued','processing','processing_media') ORDER BY id ASC",
            Database::table('jobs')
        ), ARRAY_A) ?: [];
        foreach ($rows as &$row) {
            $row['payload'] = json_decode((string) ($row['payload'] ?? ''), true) ?: [];
        }
        return $rows;
    }

    public function publicationBlockers(): array
    {
        $rows = $this->db->get_results($this->db->prepare(
            "SELECT * FROM %i WHERE status IN ('queued','processing','processing_media','paused') ORDER BY id ASC",
            Database::table('jobs')
        ), ARRAY_A) ?: [];
        foreach ($rows as &$row) {
            $row['payload'] = json_decode((string) ($row['payload'] ?? ''), true) ?: [];
        }
        return $rows;
    }

    public function updateJob(int $id, array $data): void
    {
        if (isset($data['payload']) && is_array($data['payload'])) {
            $data['payload'] = wp_json_encode($data['payload']);
        }
        $data['updated_at'] = current_time('mysql', true);
        $this->db->update(Database::table('jobs'), $data, ['id' => $id]);
    }

    /**
     * Update a job only while it is still in one of the expected states.
     *
     * This keeps a worker that is finishing a batch from overwriting a
     * concurrent cancellation with processing_media, completed, or failed.
     *
     * @param array<string,mixed> $data
     * @param array<int,string> $statuses
     */
    public function updateJobWhenStatus(int $id, array $data, array $statuses): bool
    {
        $allowedColumns = [
            'job_type', 'run_id', 'status', 'stage', 'total', 'processed',
            'succeeded', 'failed', 'media_total', 'media_processed',
            'media_failed', 'payload', 'error', 'created_by', 'started_at',
            'finished_at', 'created_at', 'updated_at',
        ];
        $statuses = array_values(array_unique(array_filter(array_map('strval', $statuses))));
        if (!$statuses) {
            return false;
        }
        if (isset($data['payload']) && is_array($data['payload'])) {
            $data['payload'] = wp_json_encode($data['payload']);
        }
        $data['updated_at'] = current_time('mysql', true);

        $assignments = [];
        $values = [Database::table('jobs')];
        foreach ($data as $column => $value) {
            if (!in_array($column, $allowedColumns, true)) {
                throw new RuntimeException('Unsupported publication job field.');
            }
            if ($value === null) {
                $assignments[] = '%i=NULL';
                $values[] = $column;
            } elseif (is_int($value) || is_bool($value)) {
                $assignments[] = '%i=%d';
                $values[] = $column;
                $values[] = (int) $value;
            } else {
                $assignments[] = '%i=%s';
                $values[] = $column;
                $values[] = (string) $value;
            }
        }
        $statusPlaceholders = implode(',', array_fill(0, count($statuses), '%s'));
        $values[] = $id;
        array_push($values, ...$statuses);
        $sql = 'UPDATE %i SET ' . implode(',', $assignments)
            . " WHERE id=%d AND status IN ({$statusPlaceholders})";
        $updated = $this->db->query($this->db->prepare($sql, ...$values));
        if ($updated === false) {
            return false;
        }
        if ($updated > 0) {
            return true;
        }

        // MySQL reports zero changed rows when every assigned value already
        // matches. Distinguish that harmless no-op from a concurrent state
        // transition such as cancellation.
        $targetStatus = isset($data['status']) ? (string) $data['status'] : null;
        return $targetStatus !== null
            && $this->job($id)['status'] === $targetStatus
            && in_array($targetStatus, $statuses, true);
    }

    public function recordWorkerScheduled(int $jobId, string $stage, float $dueAt): void
    {
        $job = $this->job($jobId);
        $payload = $job['payload'];
        $timings = (array) ($payload['timings'] ?? []);
        $scheduledAt = microtime(true);
        $timings['pending_workers'][$stage] = [
            'scheduled_at' => $scheduledAt,
            'due_at' => $dueAt,
        ];
        $timings[$stage . '_last_scheduled_at'] = gmdate('c', (int) $scheduledAt);
        $payload['timings'] = $timings;
        $this->updateJob($jobId, ['payload' => $payload]);
    }

    public function recordWorkerStarted(int $jobId, string $stage): void
    {
        $job = $this->job($jobId);
        $payload = $job['payload'];
        $timings = (array) ($payload['timings'] ?? []);
        $pending = (array) (($timings['pending_workers'] ?? [])[$stage] ?? []);
        $startedAt = microtime(true);
        $scheduledAt = (float) ($pending['scheduled_at'] ?? $startedAt);
        $dueAt = (float) ($pending['due_at'] ?? $scheduledAt);
        $schedulerWait = max(0, $startedAt - $dueAt);
        $queueTime = max(0, $startedAt - $scheduledAt);
        $timings[$stage . '_worker_passes'] = (int) ($timings[$stage . '_worker_passes'] ?? 0) + 1;
        $timings[$stage . '_scheduler_wait_seconds'] = round(
            (float) ($timings[$stage . '_scheduler_wait_seconds'] ?? 0) + $schedulerWait,
            4
        );
        $timings[$stage . '_queue_seconds'] = round(
            (float) ($timings[$stage . '_queue_seconds'] ?? 0) + $queueTime,
            4
        );
        $timings[$stage . '_last_scheduler_wait_seconds'] = round($schedulerWait, 4);
        $timings[$stage . '_max_scheduler_wait_seconds'] = round(max(
            (float) ($timings[$stage . '_max_scheduler_wait_seconds'] ?? 0),
            $schedulerWait
        ), 4);
        $timings[$stage . '_last_queue_seconds'] = round($queueTime, 4);
        $timings[$stage . '_last_started_at'] = gmdate('c', (int) $startedAt);
        unset($timings['pending_workers'][$stage]);
        $payload['timings'] = $timings;
        $this->updateJob($jobId, ['payload' => $payload]);
    }

    /** @param array<int,array<string,float|bool>> $samples */
    public function recordMediaTimings(int $jobId, array $samples): void
    {
        if (!$samples) {
            return;
        }
        $job = $this->job($jobId);
        $payload = $job['payload'];
        $timings = (array) ($payload['timings'] ?? []);
        $media = (array) ($timings['media'] ?? []);
        foreach ($samples as $sample) {
            $media['attempts'] = (int) ($media['attempts'] ?? 0) + 1;
            $media['cache_hits'] = (int) ($media['cache_hits'] ?? 0) + (!empty($sample['cache_hit']) ? 1 : 0);
            foreach (['cache_lookup_seconds', 'download_seconds', 'attachment_seconds', 'resize_seconds', 'total_seconds'] as $key) {
                $media[$key] = round((float) ($media[$key] ?? 0) + (float) ($sample[$key] ?? 0), 4);
            }
            $media['last'] = $sample;
        }
        $timings['media'] = $media;
        $payload['timings'] = $timings;
        $this->updateJob($jobId, ['payload' => $payload]);
    }

    public function completeRemoteReleaseForRun(?int $runId, string $status = 'completed'): void
    {
        if (!$runId) {
            return;
        }
        if (!in_array($status, ['completed', 'failed'], true)) {
            throw new RuntimeException('A remote release can finish as completed or failed.');
        }
        $this->db->update(Database::table('remote_releases'), [
            'status' => $status,
            'updated_at' => current_time('mysql', true),
        ], [
            'run_id' => $runId,
            'status' => 'queued',
        ]);
    }

    /** @return array{processed:int,succeeded:int,failed:int} */
    public function jobChangeCounts(array $job): array
    {
        if (($job['payload']['mode'] ?? '') === 'selected') {
            $ids = array_values(array_unique(array_filter(array_map('intval', (array) ($job['payload']['ids'] ?? [])))));
            if (!$ids) {
                return ['processed' => 0, 'succeeded' => 0, 'failed' => 0];
            }
            $placeholders = implode(',', array_fill(0, count($ids), '%d'));
            $sql = $this->db->prepare(
                'SELECT '
                . "SUM(status IN ('published','failed')) processed, "
                . "SUM(status='published') succeeded, "
                . "SUM(status='failed') failed "
                . "FROM %i WHERE id IN ({$placeholders})",
                Database::table('changes'),
                ...$ids
            );
        } else {
            $sql = $this->db->prepare(
                'SELECT '
                . "SUM(status IN ('published','failed')) processed, "
                . "SUM(status='published') succeeded, "
                . "SUM(status='failed') failed "
                . 'FROM %i WHERE run_id=%d',
                Database::table('changes'),
                (int) $job['run_id']
            );
        }
        $row = $this->db->get_row($sql, ARRAY_A) ?: [];
        return [
            'processed' => (int) ($row['processed'] ?? 0),
            'succeeded' => (int) ($row['succeeded'] ?? 0),
            'failed' => (int) ($row['failed'] ?? 0),
        ];
    }

    /** @return array<int,int> */
    public function eligibleJobChanges(array $job, int $limit): array
    {
        $limit = max(1, min(100, $limit));
        if (($job['payload']['mode'] ?? '') === 'selected') {
            $ids = array_values(array_unique(array_filter(array_map('intval', (array) ($job['payload']['ids'] ?? [])))));
            if (!$ids) {
                return [];
            }
            $placeholders = implode(',', array_fill(0, count($ids), '%d'));
            $args = array_merge([Database::table('changes')], $ids, [$limit]);
            $sql = $this->db->prepare(
                "SELECT id FROM %i WHERE status IN ('approved','auto_approved','publishing')"
                . " AND id IN ({$placeholders}) ORDER BY priority ASC, id ASC LIMIT %d",
                ...$args
            );
        } else {
            $sql = $this->db->prepare(
                "SELECT id FROM %i WHERE status IN ('approved','auto_approved','publishing')"
                . ' AND run_id=%d ORDER BY priority ASC, id ASC LIMIT %d',
                Database::table('changes'),
                (int) $job['run_id'],
                $limit
            );
        }
        return array_map('intval', $this->db->get_col($sql) ?: []);
    }

    /** @param array<int,string> $urls */
    public function enqueueMedia(int $jobId, int $changeId, int $productId, array $urls): int
    {
        $queued = 0;
        $now = current_time('mysql', true);
        foreach (array_values(array_unique(array_filter(array_map('strval', $urls)))) as $position => $url) {
            $urlHash = hash('sha256', $url);
            $existing = $this->db->get_row($this->db->prepare(
                'SELECT * FROM %i WHERE change_id=%d AND position=%d',
                Database::table('media_queue'),
                $changeId,
                $position
            ), ARRAY_A);
            if ($existing) {
                $attachmentId = (int) ($existing['attachment_id'] ?? 0);
                if (
                    $existing['status'] === 'completed'
                    && hash_equals((string) $existing['url_hash'], $urlHash)
                    && $attachmentId > 0
                    && $this->productHasAttachment($productId, $attachmentId)
                ) {
                    continue;
                }
                if (
                    (int) $existing['job_id'] === $jobId
                    && (int) $existing['product_id'] === $productId
                    && hash_equals((string) $existing['url_hash'], $urlHash)
                    && in_array($existing['status'], ['pending', 'retry', 'processing'], true)
                ) {
                    continue;
                }
                $updated = $this->db->update(Database::table('media_queue'), [
                    'job_id' => $jobId,
                    'product_id' => $productId,
                    'source_url' => $url,
                    'url_hash' => $urlHash,
                    'status' => 'pending',
                    'attempts' => 0,
                    'available_at' => $now,
                    'attachment_id' => null,
                    'error' => null,
                    'updated_at' => $now,
                ], ['id' => (int) $existing['id']]);
                $queued += (int) ($updated !== false);
                continue;
            }
            $inserted = $this->db->query($this->db->prepare(
                'INSERT IGNORE INTO %i (job_id,change_id,product_id,source_url,url_hash,position,status,attempts,available_at,created_at,updated_at) VALUES (%d,%d,%d,%s,%s,%d,%s,0,%s,%s,%s)',
                Database::table('media_queue'),
                $jobId,
                $changeId,
                $productId,
                $url,
                $urlHash,
                $position,
                'pending',
                $now,
                $now,
                $now
            ));
            $queued += (int) ($inserted === 1);
        }
        return $queued;
    }

    public function cancelUnfinishedMedia(int $jobId): int
    {
        return (int) $this->db->query($this->db->prepare(
            "UPDATE %i SET status='cancelled',attachment_id=NULL,updated_at=%s"
            . " WHERE job_id=%d AND status IN ('pending','retry','processing')",
            Database::table('media_queue'),
            current_time('mysql', true),
            $jobId
        ));
    }

    public function nextMedia(int $jobId): ?array
    {
        $row = $this->db->get_row($this->db->prepare(
            "SELECT * FROM %i WHERE job_id=%d AND ((status IN ('pending','retry') AND available_at<=%s) OR (status='processing' AND updated_at<%s)) ORDER BY id ASC LIMIT 1",
            Database::table('media_queue'),
            $jobId,
            current_time('mysql', true),
            gmdate('Y-m-d H:i:s', time() - 5 * MINUTE_IN_SECONDS)
        ), ARRAY_A);
        return $row ?: null;
    }

    public function nextMediaBatch(int $jobId, int $limit): array
    {
        return $this->db->get_results($this->db->prepare(
            "SELECT * FROM %i WHERE job_id=%d AND ((status IN ('pending','retry') AND available_at<=%s) OR (status='processing' AND updated_at<%s)) ORDER BY id ASC LIMIT %d",
            Database::table('media_queue'),
            $jobId,
            current_time('mysql', true),
            gmdate('Y-m-d H:i:s', time() - 5 * MINUTE_IN_SECONDS),
            max(1, min(10, $limit))
        ), ARRAY_A) ?: [];
    }

    public function updateMedia(int $id, array $data): void
    {
        $data['updated_at'] = current_time('mysql', true);
        $this->db->update(Database::table('media_queue'), $data, ['id' => $id]);
    }

    /** @param array<string,mixed> $data */
    public function updateMediaForActiveJob(int $id, int $jobId, array $data): bool
    {
        $allowedColumns = ['status', 'attempts', 'available_at', 'attachment_id', 'error', 'updated_at'];
        $data['updated_at'] = current_time('mysql', true);
        $assignments = [];
        $values = [Database::table('media_queue'), Database::table('jobs')];
        foreach ($data as $column => $value) {
            if (!in_array($column, $allowedColumns, true)) {
                throw new RuntimeException('Unsupported publication media field.');
            }
            if ($value === null) {
                $assignments[] = 'm.%i=NULL';
                $values[] = $column;
            } elseif (is_int($value) || is_bool($value)) {
                $assignments[] = 'm.%i=%d';
                $values[] = $column;
                $values[] = (int) $value;
            } else {
                $assignments[] = 'm.%i=%s';
                $values[] = $column;
                $values[] = (string) $value;
            }
        }
        $values[] = $id;
        $values[] = $jobId;
        $updated = $this->db->query($this->db->prepare(
            'UPDATE %i m INNER JOIN %i j ON j.id=m.job_id SET ' . implode(',', $assignments)
            . " WHERE m.id=%d AND m.job_id=%d AND j.status='processing_media'",
            ...$values
        ));
        return $updated !== false && $updated > 0;
    }

    /** @return array<string,int> */
    public function publicationDiagnostics(): array
    {
        return [
            'finished_product_jobs_still_active' => (int) $this->db->get_var($this->db->prepare(
                "SELECT COUNT(*) FROM %i j WHERE j.status='processing' AND j.processed>=j.total"
                . " AND NOT EXISTS (SELECT 1 FROM %i m WHERE m.job_id=j.id"
                . " AND m.status IN ('pending','retry','processing'))",
                Database::table('jobs'),
                Database::table('media_queue')
            )),
            'cancelled_jobs_with_unfinished_media' => (int) $this->db->get_var($this->db->prepare(
                "SELECT COUNT(*) FROM %i m INNER JOIN %i j ON j.id=m.job_id"
                . " WHERE j.status='cancelled' AND m.status IN ('pending','retry','processing')",
                Database::table('media_queue'),
                Database::table('jobs')
            )),
        ];
    }

    private function productHasAttachment(int $productId, int $attachmentId): bool
    {
        $product = wc_get_product($productId);
        if (!$product || get_post_type($attachmentId) !== 'attachment') {
            return false;
        }
        return (int) $product->get_image_id() === $attachmentId
            || in_array($attachmentId, array_map('intval', $product->get_gallery_image_ids()), true);
    }

    /** @param array<int,string> $urls */
    public function productHasMediaUrls(int $productId, array $urls): bool
    {
        $product = wc_get_product($productId);
        if (!$product) {
            return false;
        }

        $expected = [];
        foreach (array_values(array_unique(array_filter(array_map('strval', $urls)))) as $url) {
            $url = esc_url_raw($url);
            if ($url === '') {
                return false;
            }
            $attachmentId = (int) $this->db->get_var($this->db->prepare(
                'SELECT attachment_id FROM %i WHERE url_hash=%s',
                Database::table('media_assets'),
                hash('sha256', $url)
            ));
            if ($attachmentId <= 0 || get_post_type($attachmentId) !== 'attachment') {
                return false;
            }
            $expected[] = $attachmentId;
        }

        $actual = array_values(array_filter(array_merge(
            [(int) $product->get_image_id()],
            array_map('intval', $product->get_gallery_image_ids())
        )));
        return $actual === $expected;
    }

    /** @return array<int,int> */
    public function mediaAttachments(int $changeId): array
    {
        return array_map('intval', $this->db->get_col($this->db->prepare(
            "SELECT attachment_id FROM %i WHERE change_id=%d AND status='completed' AND attachment_id IS NOT NULL ORDER BY position ASC",
            Database::table('media_queue'),
            $changeId
        )) ?: []);
    }

    public function mediaCounts(int $jobId): array
    {
        $rows = $this->db->get_results($this->db->prepare(
            'SELECT status,COUNT(*) count FROM %i WHERE job_id=%d GROUP BY status',
            Database::table('media_queue'),
            $jobId
        ), ARRAY_A) ?: [];
        $counts = ['total' => 0, 'completed' => 0, 'failed' => 0, 'waiting' => 0];
        foreach ($rows as $row) {
            $count = (int) $row['count'];
            $counts['total'] += $count;
            if ($row['status'] === 'completed') {
                $counts['completed'] += $count;
            } elseif ($row['status'] === 'failed') {
                $counts['failed'] += $count;
            } else {
                $counts['waiting'] += $count;
            }
        }
        return $counts;
    }

    public function nextMediaTimestamp(int $jobId): ?int
    {
        $value = $this->db->get_var($this->db->prepare(
            "SELECT MIN(CASE WHEN status='processing' THEN DATE_ADD(updated_at, INTERVAL 5 MINUTE) ELSE available_at END) FROM %i WHERE job_id=%d AND status IN ('pending','retry','processing')",
            Database::table('media_queue'),
            $jobId
        ));
        return $value ? strtotime((string) $value . ' UTC') : null;
    }

    public function aliases(array $filters = []): array
    {
        $sql = $this->db->prepare(
            'SELECT * FROM %i ORDER BY id DESC LIMIT 500',
            Database::table('aliases')
        );
        return $this->db->get_results($sql, ARRAY_A) ?: [];
    }

    public function resolveAlias(?int $sourceId, string $type, string $value): ?string
    {
        $normalized = self::normalizeKey($value);
        $rows = $this->aliases();
        foreach ($rows as $row) {
            if ($row['status'] !== 'approved' || $row['alias_type'] !== $type) {
                continue;
            }
            if ($row['source_id'] !== null && (int) $row['source_id'] !== $sourceId) {
                continue;
            }
            if (self::normalizeKey((string) $row['source_value']) === $normalized) {
                return (string) $row['canonical_value'];
            }
        }
        return null;
    }

    public function stageAlias(?int $sourceId, string $type, string $source, string $canonical): int
    {
        foreach ($this->aliases() as $row) {
            if ((int) ($row['source_id'] ?? 0) === (int) $sourceId && $row['alias_type'] === $type && $row['source_value'] === $source) {
                return (int) $row['id'];
            }
        }
        $this->db->insert(Database::table('aliases'), [
            'source_id' => $sourceId,
            'alias_type' => $type,
            'source_value' => $source,
            'canonical_value' => $canonical,
            'status' => 'pending',
            'created_at' => current_time('mysql', true),
            'updated_at' => current_time('mysql', true),
        ]);
        return (int) $this->db->insert_id;
    }

    public function approveAlias(int $id, string $canonical): void
    {
        $this->db->update(Database::table('aliases'), [
            'canonical_value' => sanitize_text_field($canonical),
            'status' => 'approved',
            'updated_at' => current_time('mysql', true),
        ], ['id' => $id]);
    }

    public function processMissing(int $sourceId, int $runId, int $threshold): int
    {
        $this->db->query($this->db->prepare(
            'UPDATE %i SET missing_count=missing_count+1, updated_at=%s WHERE source_id=%d AND (last_seen_run IS NULL OR last_seen_run<>%d)',
            Database::table('links'),
            current_time('mysql', true),
            $sourceId,
            $runId
        ));
        $links = $this->db->get_results($this->db->prepare(
            'SELECT * FROM %i WHERE source_id=%d AND missing_count=%d',
            Database::table('links'),
            $sourceId,
            $threshold
        ), ARRAY_A) ?: [];
        foreach ($links as $link) {
            $this->addChange($runId, null, (int) $link['product_id'], 'missing', ['missing_count' => (int) $link['missing_count']], 'pending', 3);
        }
        return count($links);
    }

    public function prune(int $retentionDays): array
    {
        $cutoff = gmdate('Y-m-d H:i:s', time() - max(1, $retentionDays) * DAY_IN_SECONDS);
        $runIds = $this->db->get_col($this->db->prepare(
            "SELECT id FROM %i WHERE finished_at IS NOT NULL AND finished_at < %s AND status IN ('completed','failed')",
            Database::table('runs'),
            $cutoff
        ));
        if (!$runIds) {
            return ['runs' => 0, 'records' => 0, 'changes' => 0];
        }
        $runIds = array_map('intval', $runIds);
        $placeholders = implode(',', array_fill(0, count($runIds), '%d'));
        $records = $this->db->query($this->db->prepare(
            "DELETE FROM %i WHERE run_id IN ({$placeholders})",
            Database::table('records'),
            ...$runIds
        ));
        $changes = $this->db->query($this->db->prepare(
            "DELETE FROM %i WHERE run_id IN ({$placeholders})",
            Database::table('changes'),
            ...$runIds
        ));
        $runs = $this->db->query($this->db->prepare(
            "DELETE FROM %i WHERE id IN ({$placeholders})",
            Database::table('runs'),
            ...$runIds
        ));
        return ['runs' => (int) $runs, 'records' => (int) $records, 'changes' => (int) $changes];
    }

    private function hydrateSource(array $row): array
    {
        $row['config'] = Crypto::decrypt((string) ($row['config'] ?? ''));
        return $row;
    }

    private function hydrateProfile(array $row): array
    {
        $row['mapping'] = json_decode((string) $row['mapping'], true) ?: [];
        $row['ownership'] = json_decode((string) ($row['ownership'] ?? ''), true) ?: [];
        return $row;
    }

    private static function normalizeKey(string $value): string
    {
        $value = class_exists('Normalizer') ? (\Normalizer::normalize($value, \Normalizer::FORM_KC) ?: $value) : $value;
        return mb_strtolower(trim(preg_replace('/[\s\x{2010}-\x{2015}]+/u', ' ', $value) ?: $value));
    }
}