import { Inject, Injectable } from '@nestjs/common';
import type { CollectedProductPayload } from '@atlas/contracts';
import type { Database } from '@atlas/database';
import { ApiError } from '../common/api-error.js';
import { IdempotencyService } from '../common/idempotency.service.js';
import { DATABASE } from '../database/database.module.js';

@Injectable()
export class ExtensionService {
    public constructor(
        @Inject(DATABASE) private readonly database: Database,
        @Inject(IdempotencyService)
        private readonly idempotency: IdempotencyService,
    ) {}

    public async ingest(
        workspaceId: string,
        userId: string,
        key: string,
        payload: CollectedProductPayload,
    ) {
        return this.idempotency.run(
            {
                workspaceId,
                userId,
                method: 'POST',
                path: '/api/extension/materials',
                key,
                request: payload,
            },
            async (tx) => {
                const platform = await tx.query<{ id: string; code: string }>(
                    `SELECT id, code FROM atlas.platforms
              WHERE code = $1 AND status = 'active' AND coalesce((capabilities->>'can_be_source')::boolean, false)`,
                    [payload.source.platform.toUpperCase()],
                );
                if (!platform.rows[0])
                    throw new ApiError(
                        422,
                        'SOURCE_PLATFORM_UNAVAILABLE',
                        '该来源平台当前不可采集',
                    );
                const existing = await this.findExistingSourceItem(
                    tx,
                    workspaceId,
                    platform.rows[0].id,
                    payload,
                );
                if (existing) {
                    await this.replaceSourceSnapshot(
                        tx,
                        workspaceId,
                        existing.id,
                        existing.product_id,
                        payload,
                    );
                    return {
                        status: 200,
                        body: {
                            data: await this.materialResponse(
                                tx,
                                existing.id,
                                existing.product_id,
                                true,
                                true,
                            ),
                        },
                    };
                }
                const { status, sourcePrice } = sourceState(payload);
                const item = await tx.query<{ id: string }>(
                    `INSERT INTO atlas.collection_items
                (workspace_id, source_kind, source_platform_id, source_item_key, source_url, source_title, source_currency_code, source_min_price, source_max_price, status, raw_evidence, normalized_payload, parser_version, collected_at)
                VALUES (current_setting('app.workspace_id')::uuid, 'platform', $1, $2, $3, $4, $5, $6, $6, $7::atlas.collection_item_status, $8, $9, $10, $11) RETURNING id`,
                    [
                        platform.rows[0].id,
                        payload.source.itemKey,
                        payload.source.canonicalUrl,
                        payload.common.title?.value ?? null,
                        sourcePrice?.currencyCode ?? null,
                        sourcePrice?.amount ?? null,
                        status,
                        JSON.stringify(payload.evidence),
                        JSON.stringify(payload),
                        payload.source.parserVersion,
                        payload.source.collectedAt,
                    ],
                );
                const product = await tx.query<{ id: string }>(
                    `INSERT INTO atlas.products (workspace_id, collection_item_id, source_title, purchase_url)
                VALUES (current_setting('app.workspace_id')::uuid, $1, $2, $3) RETURNING id`,
                    [
                        item.rows[0].id,
                        payload.common.title?.value ?? null,
                        payload.source.canonicalUrl,
                    ],
                );
                await this.writeSourceSnapshotRows(
                    tx,
                    workspaceId,
                    item.rows[0].id,
                    product.rows[0].id,
                    1,
                    payload,
                    false,
                );
                return {
                    status: 201,
                    body: {
                        data: await this.materialResponse(
                            tx,
                            item.rows[0].id,
                            product.rows[0].id,
                            false,
                            false,
                        ),
                    },
                };
            },
        );
    }

    /**
     * Ozon has no public, stable SPU ID. The entered URL SKU remains the source
     * identity for audit purposes, while a re-collection from any sibling URL
     * reuses the same item when its real source SKU set intersects this one.
     */
    private async findExistingSourceItem(
        tx: { query: Database['query'] },
        workspaceId: string,
        platformId: string,
        payload: CollectedProductPayload,
    ): Promise<{ id: string; product_id: string } | null> {
        const bySourceKey = await tx.query<{ id: string; product_id: string }>(
            `SELECT c.id, p.id AS product_id
             FROM atlas.collection_items c
             JOIN atlas.products p ON p.collection_item_id = c.id
             WHERE c.source_platform_id = $1
               AND c.source_item_key = $2
               AND c.deleted_at IS NULL`,
            [platformId, payload.source.itemKey],
        );
        if (bySourceKey.rows[0]) return bySourceKey.rows[0];
        if (payload.source.platform !== 'ozon') return null;

        const sourceSkus = [
            ...new Set(
                payload.variants
                    .map(
                        (variant) =>
                            variant.sourceSku ?? variant.sourceVariantKey,
                    )
                    .filter((sku) => sku.trim().length > 0),
            ),
        ];
        if (!sourceSkus.length) return null;
        // Different sibling URLs have different source_item_key values. Locking
        // each real SKU in sorted order makes two concurrent first collections
        // of the same public variant group observe one another's snapshot.
        for (const sourceSku of sourceSkus.sort()) {
            await tx.query(
                `SELECT pg_advisory_xact_lock(hashtextextended($1::text, 0))`,
                [`${workspaceId}:${platformId}:${sourceSku}`],
            );
        }
        const byVariantOverlap = await tx.query<{
            id: string;
            product_id: string;
        }>(
            `SELECT DISTINCT c.id, p.id AS product_id
             FROM atlas.collection_items c
             JOIN atlas.products p ON p.collection_item_id = c.id
             JOIN atlas.product_variants v ON v.product_id = p.id
             WHERE c.source_platform_id = $1
               AND c.deleted_at IS NULL
               AND v.deleted_at IS NULL
               AND coalesce(v.source_sku, v.source_variant_key) = ANY($2::text[])
             LIMIT 2`,
            [platformId, sourceSkus],
        );
        if (byVariantOverlap.rows.length > 1) {
            throw new ApiError(
                409,
                'SOURCE_VARIANT_GROUP_AMBIGUOUS',
                '发现多条历史采集记录包含相同 Ozon SKU，请先在采集箱处理重复记录',
            );
        }
        return byVariantOverlap.rows[0] ?? null;
    }

    private async replaceSourceSnapshot(
        tx: { query: Database['query'] },
        workspaceId: string,
        collectionItemId: string,
        productId: string,
        payload: CollectedProductPayload,
    ) {
        const { status, sourcePrice } = sourceState(payload);
        // Once a platform profile exists, product_variants are no longer a
        // disposable collection projection: they may carry user edits and are
        // referenced by platform listings. A re-collection refreshes the
        // source evidence/candidates, but must not overwrite that draft or
        // sever a SKU's publishing history.
        const profileCount = await tx.query<{ count: string }>(
            `SELECT count(*)::text AS count
             FROM atlas.product_platform_profiles
             WHERE product_id = $1`,
            [productId],
        );
        const preserveEditorDraft =
            Number(profileCount.rows[0]?.count ?? 0) > 0;
        // 不删除旧候选/任务记录：运行中的 Worker 可能已拿到它。取消并保留审计记录，
        // 新快照在查询和 Worker 中均按 source_snapshot_revision 隔离。
        await tx.query(
            `UPDATE atlas.collection_asset_transfers SET status = 'cancelled', completed_at = now(), updated_at = now()
          WHERE collection_item_id = $1 AND status IN ('pending_selection', 'queued', 'running')`,
            [collectionItemId],
        );
        const item = await tx.query<{
            source_snapshot_revision: string | number;
        }>(
            `UPDATE atlas.collection_items
          SET source_url = $1, source_title = $2, source_currency_code = $3, source_min_price = $4, source_max_price = $4,
              status = $5::atlas.collection_item_status, raw_evidence = $6, normalized_payload = $7, parser_version = $8,
              collected_at = $9, source_snapshot_revision = source_snapshot_revision + 1, revision = revision + 1, updated_at = now()
          WHERE id = $10 AND deleted_at IS NULL RETURNING source_snapshot_revision`,
            [
                payload.source.canonicalUrl,
                payload.common.title?.value ?? null,
                sourcePrice?.currencyCode ?? null,
                sourcePrice?.amount ?? null,
                status,
                JSON.stringify(payload.evidence),
                JSON.stringify(payload),
                payload.source.parserVersion,
                payload.source.collectedAt,
                collectionItemId,
            ],
        );
        const snapshotRevision = Number(item.rows[0]?.source_snapshot_revision);
        if (!Number.isSafeInteger(snapshotRevision) || snapshotRevision < 1)
            throw new Error('采集来源快照版本更新失败');
        if (!preserveEditorDraft) {
            await tx.query(
                `UPDATE atlas.products
                 SET source_title = $1, updated_at = now(), revision = revision + 1
                 WHERE id = $2`,
                [payload.common.title?.value ?? null, productId],
            );
        }
        await this.writeSourceSnapshotRows(
            tx,
            workspaceId,
            collectionItemId,
            productId,
            snapshotRevision,
            payload,
            preserveEditorDraft,
        );
        return snapshotRevision;
    }

    private async writeSourceSnapshotRows(
        tx: { query: Database['query'] },
        workspaceId: string,
        collectionItemId: string,
        productId: string,
        snapshotRevision: number,
        payload: CollectedProductPayload,
        preserveEditorDraft: boolean,
    ) {
        const verifiedVariants = payload.variants.filter(
            (variant) => variant.sourceVariantKey.trim().length > 0,
        );
        const variantRows = verifiedVariants.length
            ? verifiedVariants
            : [
                  {
                      sourceVariantKey: null,
                      sourceSku: undefined,
                      displayName: '默认 SKU',
                  },
              ];
        const sourceKeys = variantRows
            .map((variant) => variant.sourceVariantKey)
            .filter((key): key is string => !!key);
        const existing = await tx.query<{
            id: string;
            source_variant_key: string | null;
            deleted_at: Date | string | null;
        }>(
            `SELECT id, source_variant_key, deleted_at
             FROM atlas.product_variants
             WHERE product_id = $1
               AND (source_variant_key = ANY($2::text[]) OR source_variant_key IS NULL)
             ORDER BY sort_order, id`,
            [productId, sourceKeys],
        );
        const existingBySourceKey = new Map(
            existing.rows
                .filter((variant) => variant.source_variant_key)
                .map((variant) => [variant.source_variant_key!, variant]),
        );
        // A capture without a stable source key has exactly one synthetic
        // default SKU. Reuse an existing unkeyed row when possible; this keeps
        // its platform mapping rather than attempting a duplicate insert.
        const existingDefault = existing.rows.find(
            (variant) => variant.source_variant_key === null,
        );
        const variants = new Map<string, string>();
        const retainedVariantIds: string[] = [];
        for (let index = 0; index < variantRows.length; index += 1) {
            const variant = variantRows[index];
            const matched = variant.sourceVariantKey
                ? existingBySourceKey.get(variant.sourceVariantKey)
                : existingDefault;
            let variantId: string;
            if (matched) {
                variantId = matched.id;
                // A user may have deliberately removed an Atlas SKU after
                // editing began. Keep that decision; its source evidence is
                // still present in normalized_payload, but it is not silently
                // revived into the editable/publishable SKU set.
                if (!preserveEditorDraft) {
                    await tx.query(
                        `UPDATE atlas.product_variants
                         SET source_sku = $1, display_name = $2, sort_order = $3,
                             purchase_cost_cny = COALESCE(ROUND($4::numeric, 2), purchase_cost_cny),
                             deleted_at = NULL, updated_at = now()
                         WHERE id = $5`,
                        [
                            variant.sourceSku ?? null,
                            variant.displayName,
                            index,
                            sourcePurchaseCost(payload.source.platform, variant),
                            variantId,
                        ],
                    );
                }
            } else {
                const inserted = await tx.query<{ id: string }>(
                    `INSERT INTO atlas.product_variants
                     (workspace_id, product_id, source_variant_key, source_sku, display_name, sort_order, purchase_cost_cny)
                     VALUES ($1, $2, $3, $4, $5, $6, ROUND($7::numeric, 2)) RETURNING id`,
                    [
                        workspaceId,
                        productId,
                        variant.sourceVariantKey,
                        variant.sourceSku ?? null,
                        variant.displayName,
                        index,
                        sourcePurchaseCost(payload.source.platform, variant),
                    ],
                );
                variantId = inserted.rows[0].id;
                await this.ensureProfileListingsForVariant(
                    tx,
                    productId,
                    variantId,
                );
            }
            retainedVariantIds.push(variantId);
            if (variant.sourceVariantKey)
                variants.set(variant.sourceVariantKey, variantId);
        }
        // Before an editor profile exists, variants are still just the latest
        // source projection, so a missing source SKU can be safely hidden. Do
        // not physically delete it: the partial source-key unique index does
        // not exclude soft-deleted rows, and a later refresh may legitimately
        // rediscover the same stable key. After editing begins, omission from a
        // source refresh must never act as a user-requested SKU deletion.
        if (!preserveEditorDraft) {
            await tx.query(
                `UPDATE atlas.product_variants
                 SET deleted_at = now(), updated_at = now()
                 WHERE product_id = $1 AND deleted_at IS NULL
                   AND NOT (id = ANY($2::uuid[]))`,
                [productId, retainedVariantIds],
            );
        }
        for (const asset of payload.assets) {
            await tx.query(
                `INSERT INTO atlas.collection_asset_transfers (workspace_id, collection_item_id, product_id, source_variant_id, media_type, poster_url, role, source_url, status, error_payload, source_snapshot_revision)
                VALUES ($1, $2, $3, $4, $5::atlas.collection_media_type, $6, $7::atlas.media_candidate_role, $8, 'pending_selection', $9, $10)`,
                [
                    workspaceId,
                    collectionItemId,
                    productId,
                    asset.variantSourceKey
                        ? (variants.get(asset.variantSourceKey) ?? null)
                        : null,
                    asset.mediaType,
                    asset.posterUrl ?? null,
                    asset.role,
                    asset.url,
                    JSON.stringify({
                        relation: asset.relation,
                        referer_origin: asset.refererOrigin ?? null,
                        evidence_ref: asset.evidenceRef ?? null,
                        sort_order: asset.sortOrder,
                    }),
                    snapshotRevision,
                ],
            );
        }
        // Collecting records source-media candidates only. A later explicit
        // selection queues the media and creates the materialisation event.
        await tx.query(
            `UPDATE atlas.collection_items SET status = CASE WHEN status = 'needs_review' THEN status ELSE 'ready_for_edit' END, updated_at = now() WHERE id = $1 AND source_snapshot_revision = $2`,
            [collectionItemId, snapshotRevision],
        );
    }

    private async ensureProfileListingsForVariant(
        tx: { query: Database['query'] },
        productId: string,
        variantId: string,
    ) {
        // A source refresh may discover a new real SKU after a user has
        // already entered one or more platform editors. Every existing profile
        // needs a fresh, explicitly not-listed mapping for that SKU; imported
        // mappings for other SKU rows are untouched.
        await tx.query(
            `INSERT INTO atlas.product_variant_platform_listings
             (workspace_id, product_id, profile_id, platform_id, product_variant_id, store_id, currency_code)
             SELECT current_setting('app.workspace_id')::uuid, $1, p.id, p.platform_id,
                    $2, p.target_store_id, s.currency_code
             FROM atlas.product_platform_profiles p
             JOIN atlas.stores s ON s.id = p.target_store_id
             WHERE p.product_id = $1
             ON CONFLICT (workspace_id, profile_id, product_variant_id) DO NOTHING`,
            [productId, variantId],
        );
    }

    private async materialResponse(
        tx: { query: Database['query'] },
        collectionItemId: string,
        productId: string,
        reused: boolean,
        refreshed: boolean,
    ) {
        const count = await tx.query<{
            total: string;
            pending_selection: string;
            queued: string;
            succeeded: string;
            failed: string;
        }>(
            `SELECT count(*)::text AS total, count(*) FILTER (WHERE t.status = 'pending_selection')::text AS pending_selection, count(*) FILTER (WHERE t.status = 'queued')::text AS queued, count(*) FILTER (WHERE t.status = 'succeeded')::text AS succeeded, count(*) FILTER (WHERE t.status = 'failed')::text AS failed FROM atlas.collection_asset_transfers t JOIN atlas.collection_items c ON c.id = t.collection_item_id WHERE t.collection_item_id = $1 AND t.source_snapshot_revision = c.source_snapshot_revision`,
            [collectionItemId],
        );
        const status = await tx.query<{
            status: string;
            source_title: string | null;
            variant_count: string;
        }>(
            `SELECT status, source_title,
                    (SELECT count(*)::text FROM atlas.product_variants WHERE product_id=$2 AND deleted_at IS NULL) AS variant_count
             FROM atlas.collection_items WHERE id=$1`,
            [collectionItemId, productId],
        );
        return {
            collection_item: {
                id: collectionItemId,
                status: status.rows[0]?.status ?? 'received',
                source_title: status.rows[0]?.source_title ?? null,
                variant_count: Number(status.rows[0]?.variant_count ?? 0),
            },
            product_id: productId,
            reused,
            refreshed,
            editor_entry_path: `/products/${productId}/editor?collection_item_id=${collectionItemId}`,
            materialization: {
                total: Number(count.rows[0]?.total ?? 0),
                pending_selection: Number(
                    count.rows[0]?.pending_selection ?? 0,
                ),
                queued: Number(count.rows[0]?.queued ?? 0),
                succeeded: Number(count.rows[0]?.succeeded ?? 0),
                failed: Number(count.rows[0]?.failed ?? 0),
            },
        };
    }
}

function sourceState(payload: CollectedProductPayload) {
    const verifiedVariants = payload.variants.filter(
        (variant) => variant.sourceVariantKey.trim().length > 0,
    );
    return {
        status:
            verifiedVariants.length || !payload.variantAxes.length
                ? 'ready_for_edit'
                : 'needs_review',
        sourcePrice: payload.common.sourcePrice,
    };
}

/** 仅将 1688 明确标为人民币的逐 SKU 来源价写入采购价。 */
function sourcePurchaseCost(
    platform: CollectedProductPayload['source']['platform'],
    variant: CollectedProductPayload['variants'][number] | { sourceVariantKey: null },
) {
    return platform === '1688' && 'sourcePrice' in variant && variant.sourcePrice?.currencyCode === 'CNY'
        ? variant.sourcePrice.amount
        : null;
}