import type { Database, DatabaseTransaction } from '@atlas/database';

type SourceRow = { revision: number };
type BatchItem = {
    batch_id: string;
    category_id: string;
    source_revision: number;
    priority: 'high' | 'low';
};

const platformActiveRunLimit = 6;

/** 每周只自动刷新树；类型属性仅从管理员显式创建的批次中限量领取。 */
export class PlatformCatalogScheduler {
    public constructor(private readonly database: Database) {}

    public async run() {
        const sources = await this.database.query<{ platform_id: string }>(
            `SELECT source.platform_id FROM atlas.platform_catalog_sources source
             JOIN atlas.platforms platform ON platform.id = source.platform_id
             WHERE platform.status = 'active' AND upper(platform.code) = 'OZON'`,
        );
        for (const source of sources.rows) {
            await this.schedulePlatform(source.platform_id);
        }
    }

    private async schedulePlatform(platformId: string) {
        await this.database.transaction(async (tx) => {
            const source = await tx.query<SourceRow>(
                `SELECT revision FROM atlas.platform_catalog_sources
                 WHERE platform_id = $1 FOR UPDATE`,
                [platformId],
            );
            if (!source.rows[0]) return;
            await refreshBatchStatuses(tx, platformId, source.rows[0].revision);

            const pending = await tx.query<{ count: string }>(
                `SELECT count(*)::text AS count FROM atlas.platform_catalog_sync_runs
                 WHERE platform_id = $1 AND status IN ('queued', 'running')`,
                [platformId],
            );
            let capacity = platformActiveRunLimit - Number(pending.rows[0]?.count ?? 0);
            if (capacity <= 0) return;

            if (await treeNeedsRefresh(tx, platformId)) {
                await enqueueRun(tx, {
                    platformId,
                    scopeKey: 'tree',
                    sourceRevision: source.rows[0].revision,
                    trigger: 'weekly',
                    priority: 'low',
                });
                capacity -= 1;
            }
            if (capacity <= 0) return;

            const items = await pendingBatchItems(tx, platformId, capacity);
            for (const item of items) {
                const runId = await enqueueRun(tx, {
                    platformId,
                    scopeKey: `attributes:${item.category_id}`,
                    categoryId: item.category_id,
                    sourceRevision: item.source_revision,
                    trigger: 'admin_batch',
                    priority: item.priority,
                });
                await tx.query(
                    `UPDATE atlas.platform_catalog_attribute_batch_items
                     SET run_id = $3, linked_at = now()
                     WHERE batch_id = $1 AND category_id = $2 AND run_id IS NULL`,
                    [item.batch_id, item.category_id, runId],
                );
            }
            if (items.length) {
                await tx.query(
                    `UPDATE atlas.platform_catalog_attribute_batches
                     SET status = 'running', started_at = COALESCE(started_at, now())
                     WHERE id = $1 AND status = 'queued'`,
                    [items[0].batch_id],
                );
            }
        });
    }
}

async function treeNeedsRefresh(tx: DatabaseTransaction, platformId: string) {
    const result = await tx.query<{ due: boolean }>(
        `SELECT
             (state.last_success_at IS NULL
              OR state.last_success_at <= now() - interval '7 days')
             AND NOT EXISTS (
                 SELECT 1 FROM atlas.platform_catalog_sync_runs run
                 WHERE run.platform_id = $1 AND run.scope_key = 'tree'
                   AND run.status IN ('queued', 'running')
             )
             AND NOT EXISTS (
                 SELECT 1 FROM atlas.platform_catalog_sync_runs run
                 WHERE run.platform_id = $1 AND run.scope_key = 'tree'
                   AND run.created_at > now() - interval '1 hour'
             ) AS due
         FROM (SELECT 1) marker
         LEFT JOIN atlas.platform_catalog_resource_states state
           ON state.platform_id = $1 AND state.scope_key = 'tree'`,
        [platformId],
    );
    return result.rows[0]?.due === true;
}

async function pendingBatchItems(
    tx: DatabaseTransaction,
    platformId: string,
    limit: number,
): Promise<BatchItem[]> {
    const result = await tx.query<BatchItem>(
        `SELECT item.batch_id, item.category_id, batch.source_revision,
                CASE WHEN state.last_success_at IS NULL
                           OR state.last_success_at < now() -
                              make_interval(days => batch.attribute_expire_days)
                     THEN 'high' ELSE 'low' END AS priority
         FROM atlas.platform_catalog_attribute_batches batch
         JOIN atlas.platform_catalog_attribute_batch_items item
           ON item.batch_id = batch.id AND item.platform_id = batch.platform_id
         LEFT JOIN atlas.platform_catalog_resource_states state
           ON state.platform_id = batch.platform_id
          AND state.scope_key = 'attributes:' || item.category_id
         WHERE batch.platform_id = $1 AND batch.status IN ('queued', 'running')
           AND item.run_id IS NULL
         ORDER BY CASE WHEN state.last_success_at IS NULL
                            OR state.last_success_at < now() -
                               make_interval(days => batch.attribute_expire_days)
                       THEN 0 ELSE 1 END,
                  item.created_at, item.category_id
         LIMIT $2
         FOR UPDATE OF item SKIP LOCKED`,
        [platformId, limit],
    );
    return result.rows;
}

async function refreshBatchStatuses(
    tx: DatabaseTransaction,
    platformId: string,
    sourceRevision: number,
) {
    await tx.query(
        `UPDATE atlas.platform_catalog_attribute_batches
         SET status = 'cancelled', completed_at = now()
         WHERE platform_id = $1 AND status IN ('queued', 'running')
           AND source_revision <> $2`,
        [platformId, sourceRevision],
    );
    await tx.query(
        `WITH progress AS (
             SELECT batch.id,
                    count(*) FILTER (WHERE item.run_id IS NULL) AS unlinked_count,
                    count(*) FILTER (WHERE run.status IN ('queued', 'running')) AS active_count,
                    count(*) FILTER (WHERE run.status = 'failed') AS failed_count,
                    count(*) FILTER (WHERE run.status = 'cancelled') AS cancelled_count
             FROM atlas.platform_catalog_attribute_batches batch
             JOIN atlas.platform_catalog_attribute_batch_items item ON item.batch_id = batch.id
             LEFT JOIN atlas.platform_catalog_sync_runs run ON run.id = item.run_id
             WHERE batch.platform_id = $1 AND batch.status IN ('queued', 'running')
             GROUP BY batch.id
         )
         UPDATE atlas.platform_catalog_attribute_batches batch
         SET status = CASE
                 WHEN progress.unlinked_count = 0 AND progress.active_count = 0
                     THEN CASE WHEN progress.failed_count > 0 THEN 'failed'::atlas.operation_status
                               WHEN progress.cancelled_count > 0 THEN 'cancelled'::atlas.operation_status
                               ELSE 'succeeded'::atlas.operation_status END
                 ELSE batch.status
             END,
             completed_at = CASE
                 WHEN progress.unlinked_count = 0 AND progress.active_count = 0
                     THEN COALESCE(batch.completed_at, now())
                 ELSE batch.completed_at
             END
         FROM progress WHERE batch.id = progress.id`,
        [platformId],
    );
}

async function enqueueRun(
    tx: DatabaseTransaction,
    input: {
        platformId: string;
        scopeKey: string;
        categoryId?: string;
        sourceRevision: number;
        trigger: 'weekly' | 'admin_batch';
        priority: 'high' | 'low';
    },
) {
    const active = await tx.query<{ id: string; priority: 'high' | 'low' }>(
        `SELECT id, priority FROM atlas.platform_catalog_sync_runs
         WHERE platform_id = $1 AND scope_key = $2 AND status IN ('queued', 'running')
         LIMIT 1`,
        [input.platformId, input.scopeKey],
    );
    if (active.rows[0]) {
        if (input.priority === 'high' && active.rows[0].priority === 'low') {
            await tx.query(
                `UPDATE atlas.platform_catalog_sync_runs
                 SET priority = 'high', promoted_at = now()
                 WHERE id = $1 AND status = 'queued'`,
                [active.rows[0].id],
            );
        }
        return active.rows[0].id;
    }
    const inserted = await tx.query<{ id: string }>(
        `INSERT INTO atlas.platform_catalog_sync_runs
            (platform_id, scope, scope_key, category_id, trigger_kind,
             priority, source_revision)
         VALUES ($1, $2, $3, $4, $5, $6, $7)
         RETURNING id`,
        [input.platformId, input.categoryId ? 'attributes' : 'tree', input.scopeKey,
            input.categoryId ?? null, input.trigger, input.priority, input.sourceRevision],
    );
    await tx.query(
        `INSERT INTO atlas.platform_catalog_outbox_events (run_id, platform_id)
         VALUES ($1, $2)`,
        [inserted.rows[0].id, input.platformId],
    );
    return inserted.rows[0].id;
}