atlas-seller-kit/apps/api/src/platform-catalog-admin/platform-catalog-admin.reads.ts

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

@Injectable()
export class PlatformCatalogAdminReads {
    public constructor(@Inject(DATABASE) private readonly database: Database) {}

    public async platforms() {
        const rows = await this.database.query<{
            id: string; code: string; name: string; status: string;
            client_id: string | null; source_updated_at: Date | null;
            category_count: string; last_tree_sync_at: Date | null;
            next_tree_sync_at: Date | null; last_error: string | null;
        }>(
            `SELECT p.id, p.code, p.name, p.status,
                    source.client_id, source.updated_at AS source_updated_at,
                    (SELECT count(*) FROM atlas.platform_product_categories category
                     WHERE category.platform_id = p.id AND category.is_active)::text AS category_count,
                    tree_state.last_success_at AS last_tree_sync_at,
                    tree_state.last_success_at + interval '7 days' AS next_tree_sync_at,
                    tree_state.last_error_message AS last_error
             FROM atlas.platforms p
             LEFT JOIN atlas.platform_catalog_sources source ON source.platform_id = p.id
             LEFT JOIN atlas.platform_catalog_resource_states tree_state
               ON tree_state.platform_id = p.id AND tree_state.scope_key = 'tree'
             ORDER BY CASE WHEN upper(p.code) = 'OZON' THEN 0 ELSE 1 END, p.name`,
        );
        return rows.rows.map((row) => ({
            id: row.id,
            code: row.code,
            name: row.name,
            status: row.status,
            source: {
                configured: row.client_id !== null,
                client_id_hint: row.client_id ? `••••${row.client_id.slice(-4)}` : null,
                updated_at: row.source_updated_at,
            },
            summary: {
                category_count: Number(row.category_count),
                last_tree_sync_at: row.last_tree_sync_at,
                next_tree_sync_at: row.next_tree_sync_at,
                last_error: row.last_error,
            },
        }));
    }

    public async categories(platformId: string) {
        await this.requirePlatform(platformId);
        const result = await this.database.query(
            `SELECT category.id, category.parent_id, category.external_category_id,
                    category.external_type_id, category.name, category.path,
                    category.is_active, category.synced_at,
                    state.last_success_at AS attribute_last_success_at,
                    state.last_error_message AS attribute_last_error,
                    state.last_error_at AS attribute_last_error_at,
                    CASE
                        WHEN category.external_type_id IS NULL OR category.external_type_id = '0'
                            THEN NULL
                        WHEN state.last_success_at IS NULL THEN 'missing'
                        WHEN state.last_success_at > now() -
                             make_interval(days => COALESCE(policy.attribute_fresh_days, 14))
                            THEN 'fresh'
                        WHEN state.last_success_at >= now() -
                             make_interval(days => COALESCE(policy.attribute_expire_days, 21))
                            THEN 'stale'
                        ELSE 'expired'
                    END AS attribute_cache_status
             FROM atlas.platform_product_categories category
             LEFT JOIN atlas.platform_catalog_resource_states state
               ON state.platform_id = category.platform_id
              AND state.scope_key = 'attributes:' || category.id
             LEFT JOIN atlas.platform_catalog_refresh_policies policy
               ON policy.platform_id = category.platform_id
             WHERE category.platform_id = $1 ORDER BY path, id LIMIT 20000`,
            [platformId],
        );
        return result.rows;
    }

    public async refreshPolicy(platformId: string) {
        await this.requirePlatform(platformId);
        const result = await this.database.query<{
            attribute_fresh_days: number; attribute_expire_days: number;
            dictionary_fresh_days: number; dictionary_expire_days: number;
            updated_at: Date | null;
        }>(
            `SELECT COALESCE(policy.attribute_fresh_days, 14)::int AS attribute_fresh_days,
                    COALESCE(policy.attribute_expire_days, 21)::int AS attribute_expire_days,
                    COALESCE(policy.dictionary_fresh_days, 14)::int AS dictionary_fresh_days,
                    COALESCE(policy.dictionary_expire_days, 21)::int AS dictionary_expire_days,
                    policy.updated_at
             FROM atlas.platforms platform
             LEFT JOIN atlas.platform_catalog_refresh_policies policy
               ON policy.platform_id = platform.id
             WHERE platform.id = $1`,
            [platformId],
        );
        return result.rows[0];
    }

    public async attributeSyncSummary(platformId: string) {
        await this.requirePlatform(platformId);
        const result = await this.database.query<{
            total: string; fresh: string; stale: string; expired: string; missing: string;
            active_batch_id: string | null; evaluated_at: Date;
        }>(
            `WITH policy AS (
                 SELECT COALESCE(refresh.attribute_fresh_days, 14)::int AS fresh_days,
                        COALESCE(refresh.attribute_expire_days, 21)::int AS expire_days
                 FROM atlas.platforms platform
                 LEFT JOIN atlas.platform_catalog_refresh_policies refresh
                   ON refresh.platform_id = platform.id
                 WHERE platform.id = $1
             ), resources AS (
                 SELECT state.last_success_at, policy.fresh_days, policy.expire_days
                 FROM atlas.platform_product_categories category
                 CROSS JOIN policy
                 LEFT JOIN atlas.platform_catalog_resource_states state
                   ON state.platform_id = category.platform_id
                  AND state.scope_key = 'attributes:' || category.id
                 WHERE category.platform_id = $1 AND category.is_active
                   AND category.external_type_id IS NOT NULL AND category.external_type_id <> '0'
             )
             SELECT count(*)::text AS total,
                    count(*) FILTER (
                        WHERE last_success_at > now() - make_interval(days => fresh_days)
                    )::text AS fresh,
                    count(*) FILTER (
                        WHERE last_success_at <= now() - make_interval(days => fresh_days)
                          AND last_success_at >= now() - make_interval(days => expire_days)
                    )::text AS stale,
                    count(*) FILTER (
                        WHERE last_success_at < now() - make_interval(days => expire_days)
                    )::text AS expired,
                    count(*) FILTER (WHERE last_success_at IS NULL)::text AS missing,
                    (SELECT id FROM atlas.platform_catalog_attribute_batches batch
                     WHERE batch.platform_id = $1 AND batch.status IN ('queued', 'running')
                     LIMIT 1) AS active_batch_id,
                    now() AS evaluated_at
             FROM resources`,
            [platformId],
        );
        const row = result.rows[0];
        return {
            total: Number(row?.total ?? 0),
            fresh: Number(row?.fresh ?? 0),
            stale: Number(row?.stale ?? 0),
            expired: Number(row?.expired ?? 0),
            missing: Number(row?.missing ?? 0),
            active_batch_id: row?.active_batch_id ?? null,
            evaluated_at: row?.evaluated_at ?? new Date(),
        };
    }

    public async attributes(platformId: string, categoryId: string) {
        await this.requireCategory(platformId, categoryId);
        const result = await this.database.query(
            `SELECT attribute.id, attribute.platform_attribute_id, attribute.complex_id,
                    attribute.name, attribute.description, attribute.group_id, attribute.group_name,
                    attribute.value_type, attribute.is_required, attribute.is_aspect,
                    attribute.is_collection, attribute.max_value_count, attribute.dictionary_id,
                    attribute.category_dependent, attribute.is_active, attribute.synced_at,
                    schema_state.last_success_at,
                    CASE
                        WHEN schema_state.last_success_at IS NULL THEN 'missing'
                        WHEN schema_state.last_success_at > now() -
                             make_interval(days => COALESCE(policy.attribute_fresh_days, 14))
                            THEN 'fresh'
                        WHEN schema_state.last_success_at >= now() -
                             make_interval(days => COALESCE(policy.attribute_expire_days, 21))
                            THEN 'stale'
                        ELSE 'expired'
                    END AS cache_status,
                    schema_state.last_error_message,
                    schema_state.last_error_at,
                    dictionary_state.last_success_at AS dictionary_last_success_at,
                    CASE
                        WHEN attribute.dictionary_id IS NULL THEN NULL
                        WHEN dictionary_state.last_success_at IS NULL THEN 'missing'
                        WHEN dictionary_state.last_success_at > now() -
                             make_interval(days => COALESCE(policy.dictionary_fresh_days, 14))
                            THEN 'fresh'
                        WHEN dictionary_state.last_success_at >= now() -
                             make_interval(days => COALESCE(policy.dictionary_expire_days, 21))
                            THEN 'stale'
                        ELSE 'expired'
                    END AS dictionary_cache_status,
                    dictionary_state.is_complete AS dictionary_is_complete,
                    dictionary_state.last_error_message AS dictionary_last_error_message,
                    dictionary_state.last_error_at AS dictionary_last_error_at,
                    (SELECT count(*)::int FROM atlas.platform_catalog_dictionary_values value
                     WHERE value.attribute_id = attribute.id AND value.is_active) AS dictionary_value_count
             FROM atlas.platform_product_category_attributes attribute
             LEFT JOIN atlas.platform_catalog_resource_states schema_state
               ON schema_state.platform_id = attribute.platform_id
              AND schema_state.scope_key = 'attributes:' || attribute.platform_product_category_id
             LEFT JOIN atlas.platform_catalog_resource_states dictionary_state
               ON dictionary_state.platform_id = attribute.platform_id
              AND dictionary_state.scope_key = 'dictionary:' || attribute.id
             LEFT JOIN atlas.platform_catalog_refresh_policies policy
               ON policy.platform_id = attribute.platform_id
             WHERE attribute.platform_id = $1 AND attribute.platform_product_category_id = $2
             ORDER BY attribute.is_required DESC, attribute.group_name NULLS LAST,
                      attribute.name, attribute.platform_attribute_id`,
            [platformId, categoryId],
        );
        return result.rows;
    }

    public async values(
        platformId: string,
        attributeId: string,
        keyword: string | undefined,
        page: number,
    ) {
        const attribute = await this.database.query<{ dictionary_id: string | null }>(
            `SELECT dictionary_id FROM atlas.platform_product_category_attributes
             WHERE id = $1 AND platform_id = $2 AND is_active`,
            [attributeId, platformId],
        );
        if (!attribute.rows[0]) throw new ApiError(404, 'ATTRIBUTE_NOT_FOUND', '属性不存在');
        if (!attribute.rows[0].dictionary_id) {
            throw new ApiError(422, 'ATTRIBUTE_DICTIONARY_UNAVAILABLE', '该属性没有字典候选值');
        }
        const q = keyword?.trim() ?? '';
        const [items, total, lastRun] = await Promise.all([
            this.database.query(
                `SELECT id, external_value_id, value, info, picture
                 FROM atlas.platform_catalog_dictionary_values
                 WHERE attribute_id = $1 AND is_active
                   AND ($2 = '' OR value ILIKE '%' || $2 || '%' OR external_value_id = $2)
                 ORDER BY value, external_value_id LIMIT 100 OFFSET $3`,
                [attributeId, q, (page - 1) * 100],
            ),
            this.database.query<{ count: string }>(
                `SELECT count(*)::text AS count FROM atlas.platform_catalog_dictionary_values
                 WHERE attribute_id = $1 AND is_active
                   AND ($2 = '' OR value ILIKE '%' || $2 || '%' OR external_value_id = $2)`,
                [attributeId, q],
            ),
            this.database.query<{ is_complete: boolean }>(
                `SELECT is_complete FROM atlas.platform_catalog_resource_states
                 WHERE platform_id = $1 AND scope_key = $2`,
                [platformId, `dictionary:${attributeId}`],
            ),
        ]);
        return {
            data: items.rows,
            meta: {
                page,
                page_size: 100,
                total: Number(total.rows[0]?.count ?? 0),
                is_complete: lastRun.rows[0]?.is_complete ?? false,
            },
        };
    }

    public async runs(platformId: string, limit: number) {
        await this.requirePlatform(platformId);
        const result = await this.database.query(
            `SELECT id, platform_id, scope, scope_key, category_id, attribute_id,
                    trigger_kind AS trigger, priority, status, result, error_code, error_message,
                    attempt_count, created_at, started_at, completed_at
             FROM atlas.platform_catalog_sync_runs
             WHERE platform_id = $1 ORDER BY created_at DESC LIMIT $2`,
            [platformId, limit],
        );
        return result.rows;
    }

    public async attributeBatches(platformId: string, limit: number) {
        await this.requirePlatform(platformId);
        const result = await this.database.query(
            `${batchProgressSelect}
             WHERE batch.platform_id = $1
             GROUP BY batch.id
             ORDER BY batch.created_at DESC LIMIT $2`,
            [platformId, limit],
        );
        return result.rows;
    }

    public async attributeBatch(platformId: string, batchId: string) {
        await this.requirePlatform(platformId);
        const result = await this.database.query(
            `${batchProgressSelect}
             WHERE batch.platform_id = $1 AND batch.id = $2
             GROUP BY batch.id`,
            [platformId, batchId],
        );
        if (!result.rows[0]) {
            throw new ApiError(404, 'CATALOG_ATTRIBUTE_BATCH_NOT_FOUND', '类型属性同步批次不存在');
        }
        return result.rows[0];
    }

    public async requirePlatform(platformId: string) {
        const result = await this.database.query<{ code: string }>(
            'SELECT code FROM atlas.platforms WHERE id = $1',
            [platformId],
        );
        if (!result.rows[0]) throw new ApiError(404, 'PLATFORM_NOT_FOUND', '平台不存在');
        return result.rows[0];
    }

    public async requireCategory(platformId: string, categoryId: string) {
        const result = await this.database.query(
            `SELECT 1 FROM atlas.platform_product_categories WHERE id = $1 AND platform_id = $2`,
            [categoryId, platformId],
        );
        if (!result.rows[0]) throw new ApiError(404, 'CATEGORY_NOT_FOUND', '类目不属于当前平台');
    }
}

const batchProgressSelect = `
    SELECT batch.id, batch.platform_id, batch.mode, batch.status, batch.total_count,
           count(*) FILTER (WHERE item.run_id IS NULL OR run.status = 'queued')::int
               AS pending_count,
           count(*) FILTER (WHERE run.status = 'running')::int AS running_count,
           count(*) FILTER (WHERE run.status = 'succeeded')::int AS succeeded_count,
           count(*) FILTER (WHERE run.status = 'failed')::int AS failed_count,
           count(*) FILTER (WHERE run.status = 'cancelled')::int AS cancelled_count,
           batch.created_at, batch.started_at, batch.completed_at
    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`;