// Copyright The kweaver.ai Authors.
//
// Licensed under the Apache License, Version 2.0.
// See the LICENSE file in the project root for details.

// Package resource provides Resource data access operations.
package resource

import (
	"context"
	"database/sql"
	"fmt"
	"sync"

	sq "github.com/Masterminds/squirrel"
	"github.com/bytedance/sonic"
	libCommon "github.com/kweaver-ai/kweaver-go-lib/common"
	libdb "github.com/kweaver-ai/kweaver-go-lib/db"
	"github.com/kweaver-ai/kweaver-go-lib/logger"
	"github.com/kweaver-ai/kweaver-go-lib/otel/otellog"
	"github.com/kweaver-ai/kweaver-go-lib/otel/oteltrace"
	attr "go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"

	"vega-backend/common"
	"vega-backend/drivenadapters/entityextension"
	"vega-backend/interfaces"
)

const (
	RESOURCE_TABLE_NAME = "t_resource"
)

var (
	rAccessOnce sync.Once
	rAccess     interfaces.ResourceAccess
)

type resourceAccess struct {
	appSetting *common.AppSetting
	db         *sql.DB
}

// NewResourceAccess creates ra new ResourceAccess.
func NewResourceAccess(appSetting *common.AppSetting) interfaces.ResourceAccess {
	rAccessOnce.Do(func() {
		rAccess = &resourceAccess{
			appSetting: appSetting,
			db:         libdb.NewDB(&appSetting.DBSetting),
		}
	})
	return rAccess
}

// Create creates ra new Resource.
func (ra *resourceAccess) Create(ctx context.Context, resource *interfaces.Resource) error {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Insert into resource")
	defer span.End()

	span.SetAttributes(
		attr.Key("db_url").String(libdb.GetDBUrl()),
		attr.Key("db_type").String(libdb.GetDBType()))

	// tags 转成 string 的格式
	tagsStr := libCommon.TagSlice2TagString(resource.Tags)

	// 序列化 SourceMetadata, SchemaDefinition, LogicDefinition
	sourceMetadataBytes, _ := sonic.Marshal(resource.SourceMetadata)
	if resource.SourceMetadata == nil {
		sourceMetadataBytes = []byte("{}")
	}
	schemaDefinitionBytes, _ := sonic.Marshal(resource.SchemaDefinition)
	if resource.SchemaDefinition == nil {
		schemaDefinitionBytes = []byte("[]")
	}
	logicDefinitionBytes, _ := sonic.Marshal(resource.LogicDefinition)
	if resource.LogicDefinition == nil {
		logicDefinitionBytes = []byte("[]")
	}

	sqlStr, vals, err := sq.Insert(RESOURCE_TABLE_NAME).
		Columns(
			"f_id",
			"f_catalog_id",
			"f_name",
			"f_tags",
			"f_description",
			"f_category",
			"f_status",
			"f_status_message",
			"f_last_discover_status",
			"f_database",
			"f_source_identifier",
			"f_source_metadata",
			"f_schema_definition",

			"f_logic_type",
			"f_logic_definition",

			"f_local_enabled",
			"f_local_storage_engine",
			"f_local_storage_config",
			"f_local_index_name",

			"f_sync_strategy",
			"f_sync_config",
			"f_sync_status",
			"f_last_sync_time",
			"f_sync_error_message",

			"f_creator",
			"f_creator_type",
			"f_create_time",
			"f_updater",
			"f_updater_type",
			"f_update_time",
		).
		Values(
			resource.ID,
			resource.CatalogID,
			resource.Name,
			tagsStr,
			resource.Description,
			resource.Category,
			resource.Status,
			resource.StatusMessage,
			resource.LastDiscoverStatus,
			resource.Database,
			resource.SourceIdentifier,
			string(sourceMetadataBytes),
			string(schemaDefinitionBytes),

			resource.LogicType,
			string(logicDefinitionBytes),

			false,
			"",
			"",
			"",

			"",
			"",
			"",
			0,
			"",

			resource.Creator.ID,
			resource.Creator.Type,
			resource.CreateTime,
			resource.Updater.ID,
			resource.Updater.Type,
			resource.UpdateTime,
		).ToSql()
	if err != nil {
		otellog.LogError(ctx, "Failed to build insert resource sql", err)
		return err
	}

	otellog.LogInfo(ctx, fmt.Sprintf("Insert resource SQL: %s", sqlStr))

	_, err = ra.db.ExecContext(ctx, sqlStr, vals...)
	if err != nil {
		otellog.LogError(ctx, "Insert resource failed", err)
		return err
	}

	span.SetStatus(codes.Ok, "")
	return nil
}

// GetByID retrieves ra Resource by ID.
func (ra *resourceAccess) GetByID(ctx context.Context, id string) (*interfaces.Resource, error) {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Query resource by ID")
	defer span.End()

	span.SetAttributes(attr.Key("resource_id").String(id))

	sqlStr, vals, err := sq.Select(
		"f_id",
		"f_catalog_id",
		"f_name",
		"f_tags",
		"f_description",
		"f_category",
		"f_status",
		"f_status_message",
		"f_last_discover_status",
		"f_database",
		"f_source_identifier",
		"f_source_metadata",
		"f_schema_definition",
		"f_logic_type",
		"f_logic_definition",
		"f_creator",
		"f_creator_type",
		"f_create_time",
		"f_updater",
		"f_updater_type",
		"f_update_time",
		"f_local_index_name",
	).From(RESOURCE_TABLE_NAME).
		Where(sq.Eq{"f_id": id}).
		ToSql()
	if err != nil {
		logger.Errorf("Failed to build query resource sql: %v", err)
		span.SetStatus(codes.Error, "Build sql failed")
		return nil, err
	}

	resource := &interfaces.Resource{}
	var tagsStr string
	var database, sourceIdentifier, sourceMetadata, schemaDefinition, logicDefinition sql.NullString

	row := ra.db.QueryRowContext(ctx, sqlStr, vals...)
	err = row.Scan(
		&resource.ID,
		&resource.CatalogID,
		&resource.Name,
		&tagsStr,
		&resource.Description,
		&resource.Category,
		&resource.Status,
		&resource.StatusMessage,
		&resource.LastDiscoverStatus,
		&database,
		&sourceIdentifier,
		&sourceMetadata,
		&schemaDefinition,
		&resource.LogicType,
		&logicDefinition,
		&resource.Creator.ID,
		&resource.Creator.Type,
		&resource.CreateTime,
		&resource.Updater.ID,
		&resource.Updater.Type,
		&resource.UpdateTime,
		&resource.LocalIndexName,
	)
	if err == sql.ErrNoRows {
		span.SetStatus(codes.Ok, "")
		return nil, nil
	}
	if err != nil {
		logger.Errorf("Scan resource failed: %v", err)
		span.SetStatus(codes.Error, "Scan failed")
		return nil, err
	}

	// tags string 转成数组的格式
	resource.Tags = libCommon.TagString2TagSlice(tagsStr)
	resource.Database = database.String
	resource.SourceIdentifier = sourceIdentifier.String
	if sourceMetadata.Valid && sourceMetadata.String != "" {
		_ = sonic.Unmarshal([]byte(sourceMetadata.String), &resource.SourceMetadata)
	}
	if schemaDefinition.Valid && schemaDefinition.String != "" {
		_ = sonic.Unmarshal([]byte(schemaDefinition.String), &resource.SchemaDefinition)
	}
	if logicDefinition.Valid && logicDefinition.String != "" {
		_ = sonic.Unmarshal([]byte(logicDefinition.String), &resource.LogicDefinition)
	}

	if err := attachSingleResourceExtensions(ctx, ra.appSetting, resource); err != nil {
		span.SetStatus(codes.Error, "Load resource extensions failed")
		return nil, err
	}

	span.SetStatus(codes.Ok, "")
	return resource, nil
}

// GetByIDs retrieves ra Resource by IDs.
func (ra *resourceAccess) GetByIDs(ctx context.Context, ids []string) ([]*interfaces.Resource, error) {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Query resources by IDs")
	defer span.End()

	span.SetAttributes(attr.Key("resource_ids").StringSlice(ids))

	sqlStr, vals, err := sq.Select(
		"f_id",
		"f_catalog_id",
		"f_name",
		"f_tags",
		"f_description",
		"f_category",
		"f_status",
		"f_status_message",
		"f_last_discover_status",
		"f_database",
		"f_source_identifier",
		"f_source_metadata",
		"f_schema_definition",
		"f_logic_type",
		"f_logic_definition",
		"f_creator",
		"f_creator_type",
		"f_create_time",
		"f_updater",
		"f_updater_type",
		"f_update_time",
		"f_local_index_name",
	).From(RESOURCE_TABLE_NAME).
		Where(sq.Eq{"f_id": ids}).
		ToSql()
	if err != nil {
		logger.Errorf("Failed to build query resource sql: %v", err)
		span.SetStatus(codes.Error, "Build sql failed")
		return []*interfaces.Resource{}, err
	}

	rows, err := ra.db.QueryContext(ctx, sqlStr, vals...)
	if err != nil {
		logger.Errorf("Query resources failed: %v", err)
		span.SetStatus(codes.Error, "Query failed")
		return []*interfaces.Resource{}, err
	}
	defer func() { _ = rows.Close() }()

	resources := make([]*interfaces.Resource, 0)
	for rows.Next() {
		resource := &interfaces.Resource{}
		var tagsStr string
		var database, sourceIdentifier, sourceMetadata, schemaDefinition, logicDefinition sql.NullString

		err := rows.Scan(
			&resource.ID,
			&resource.CatalogID,
			&resource.Name,
			&tagsStr,
			&resource.Description,
			&resource.Category,
			&resource.Status,
			&resource.StatusMessage,
			&resource.LastDiscoverStatus,
			&database,
			&sourceIdentifier,
			&sourceMetadata,
			&schemaDefinition,
			&resource.LogicType,
			&logicDefinition,
			&resource.Creator.ID,
			&resource.Creator.Type,
			&resource.CreateTime,
			&resource.Updater.ID,
			&resource.Updater.Type,
			&resource.UpdateTime,
			&resource.LocalIndexName,
		)

		if err != nil {
			logger.Errorf("Scan resource row failed: %v", err)
			span.SetStatus(codes.Error, "Scan row failed")
			return []*interfaces.Resource{}, err
		}

		// tags string 转成数组的格式
		resource.Tags = libCommon.TagString2TagSlice(tagsStr)
		resource.Database = database.String
		resource.SourceIdentifier = sourceIdentifier.String
		if sourceMetadata.Valid && sourceMetadata.String != "" {
			_ = sonic.Unmarshal([]byte(sourceMetadata.String), &resource.SourceMetadata)
		}
		if schemaDefinition.Valid && schemaDefinition.String != "" {
			_ = sonic.Unmarshal([]byte(schemaDefinition.String), &resource.SchemaDefinition)
		}
		if logicDefinition.Valid && logicDefinition.String != "" {
			_ = sonic.Unmarshal([]byte(logicDefinition.String), &resource.LogicDefinition)
		}

		resources = append(resources, resource)
	}

	if err := attachResourceExtensions(ctx, ra.appSetting, interfaces.ResourcesQueryParams{IncludeExtensions: false}, resources); err != nil {
		span.SetStatus(codes.Error, "Load resource extensions failed")
		return []*interfaces.Resource{}, err
	}

	span.SetStatus(codes.Ok, "")
	return resources, nil
}

// AttachListExtensions implements interfaces.ResourceAccess.
func (ra *resourceAccess) AttachListExtensions(ctx context.Context, params interfaces.ResourcesQueryParams, resources []*interfaces.Resource) error {
	return attachResourceExtensions(ctx, ra.appSetting, params, resources)
}

// GetByIDsBasic retrieves Resources by IDs without parsing sourceMetadata, schemaDefinition and logicDefinition.
// This method is optimized for memory usage when these large fields are not needed.
func (ra *resourceAccess) GetByIDsBasic(ctx context.Context, ids []string) ([]*interfaces.Resource, error) {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Query resources by IDs (basic)")
	defer span.End()

	span.SetAttributes(attr.Key("resource_ids").StringSlice(ids))

	sqlStr, vals, err := sq.Select(
		"f_id",
		"f_catalog_id",
		"f_name",
		"f_tags",
		"f_description",
		"f_category",
		"f_status",
		"f_status_message",
		"f_last_discover_status",
		"f_database",
		"f_source_identifier",
		"f_logic_type",
		"f_creator",
		"f_creator_type",
		"f_create_time",
		"f_updater",
		"f_updater_type",
		"f_update_time",
	).From(RESOURCE_TABLE_NAME).
		Where(sq.Eq{"f_id": ids}).
		ToSql()
	if err != nil {
		logger.Errorf("Failed to build query resource sql: %v", err)
		span.SetStatus(codes.Error, "Build sql failed")
		return []*interfaces.Resource{}, err
	}

	rows, err := ra.db.QueryContext(ctx, sqlStr, vals...)
	if err != nil {
		logger.Errorf("Query resources failed: %v", err)
		span.SetStatus(codes.Error, "Query failed")
		return []*interfaces.Resource{}, err
	}
	defer func() { _ = rows.Close() }()

	resources := make([]*interfaces.Resource, 0)
	for rows.Next() {
		resource := &interfaces.Resource{}
		var tagsStr string
		var database, sourceIdentifier sql.NullString

		err := rows.Scan(
			&resource.ID,
			&resource.CatalogID,
			&resource.Name,
			&tagsStr,
			&resource.Description,
			&resource.Category,
			&resource.Status,
			&resource.StatusMessage,
			&resource.LastDiscoverStatus,
			&database,
			&sourceIdentifier,
			&resource.LogicType,
			&resource.Creator.ID,
			&resource.Creator.Type,
			&resource.CreateTime,
			&resource.Updater.ID,
			&resource.Updater.Type,
			&resource.UpdateTime,
		)

		if err != nil {
			logger.Errorf("Scan resource row failed: %v", err)
			span.SetStatus(codes.Error, "Scan row failed")
			return []*interfaces.Resource{}, err
		}

		// tags string 转成数组的格式
		resource.Tags = libCommon.TagString2TagSlice(tagsStr)
		resource.Database = database.String
		resource.SourceIdentifier = sourceIdentifier.String
		// 不解析sourceMetadata、schemaDefinition和logicDefinition，以减少内存占用
		// 这些字段在需要时可以单独获取

		resources = append(resources, resource)
	}

	span.SetStatus(codes.Ok, "")
	return resources, nil
}

// GetByName retrieves ra Resource by catalog and name.
func (ra *resourceAccess) GetByName(ctx context.Context, catalogID string, name string) (*interfaces.Resource, error) {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Query resource by name")
	defer span.End()

	span.SetAttributes(attr.Key("resource_name").String(name))

	sqlStr, vals, err := sq.Select(
		"f_id",
		"f_catalog_id",
		"f_name",
		"f_tags",
		"f_description",
		"f_category",
		"f_status",
		"f_status_message",
		"f_last_discover_status",
		"f_database",
		"f_source_identifier",
		"f_source_metadata",
		"f_schema_definition",
		"f_creator",
		"f_creator_type",
		"f_create_time",
		"f_updater",
		"f_updater_type",
		"f_update_time",
	).From(RESOURCE_TABLE_NAME).
		Where(sq.Eq{"f_catalog_id": catalogID}).
		Where(sq.Eq{"f_name": name}).
		ToSql()
	if err != nil {
		logger.Errorf("Failed to build select resource sql: %v", err)
		span.SetStatus(codes.Error, "Build sql failed")
		return nil, err
	}

	resource := &interfaces.Resource{}
	var tagsStr string
	var database, sourceIdentifier, sourceMetadata, schemaDefinition sql.NullString

	row := ra.db.QueryRowContext(ctx, sqlStr, vals...)
	err = row.Scan(
		&resource.ID,
		&resource.CatalogID,
		&resource.Name,
		&tagsStr,
		&resource.Description,
		&resource.Category,
		&resource.Status,
		&resource.StatusMessage,
		&resource.LastDiscoverStatus,
		&database,
		&sourceIdentifier,
		&sourceMetadata,
		&schemaDefinition,
		&resource.Creator.ID,
		&resource.Creator.Type,
		&resource.CreateTime,
		&resource.Updater.ID,
		&resource.Updater.Type,
		&resource.UpdateTime,
	)
	if err == sql.ErrNoRows {
		span.SetStatus(codes.Ok, "")
		return nil, nil
	}
	if err != nil {
		logger.Errorf("Scan resource failed: %v", err)
		span.SetStatus(codes.Error, "Scan failed")
		return nil, err
	}

	// tags string 转成数组的格式
	resource.Tags = libCommon.TagString2TagSlice(tagsStr)
	resource.Database = database.String
	resource.SourceIdentifier = sourceIdentifier.String
	if sourceMetadata.Valid && sourceMetadata.String != "" {
		_ = sonic.Unmarshal([]byte(sourceMetadata.String), &resource.SourceMetadata)
	}
	if schemaDefinition.Valid && schemaDefinition.String != "" {
		_ = sonic.Unmarshal([]byte(schemaDefinition.String), &resource.SchemaDefinition)
	}

	span.SetStatus(codes.Ok, "")
	return resource, nil
}

// ListIDs lists Resource IDs with filters.
func (ra *resourceAccess) ListIDs(ctx context.Context, params interfaces.ResourcesQueryParams) ([]string, error) {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "List resource IDs")
	defer span.End()

	builder := sq.Select(resourceExtCol(params, "f_id")).From(RESOURCE_TABLE_NAME)

	if params.Name != "" {
		name := "%" + common.EscapeLikePattern(params.Name) + "%"
		builder = builder.Where(sq.Like{resourceExtCol(params, "f_name"): name})
	}
	if params.CatalogID != "" {
		builder = builder.Where(sq.Eq{resourceExtCol(params, "f_catalog_id"): params.CatalogID})
	}
	if params.Category != "" {
		builder = builder.Where(sq.Eq{resourceExtCol(params, "f_category"): params.Category})
	}
	if params.Status != "" {
		builder = builder.Where(sq.Eq{resourceExtCol(params, "f_status"): params.Status})
	}
	if params.Database != "" {
		builder = builder.Where(sq.Eq{resourceExtCol(params, "f_database"): params.Database})
	}

	builder = applyResourceExtensionJoins(builder, params)

	// 排序
	if params.Sort != "" {
		builder = builder.OrderBy(resourceListOrderExpr(params))
	} else {
		if len(params.ExtensionKeys) > 0 {
			builder = builder.OrderBy(fmt.Sprintf("t_resource.f_update_time %s", params.Direction))
		} else {
			builder = builder.OrderBy("f_update_time DESC")
		}
	}

	sqlStr, vals, err := builder.ToSql()
	if err != nil {
		span.SetStatus(codes.Error, "Build sql failed")
		return nil, err
	}

	rows, err := ra.db.QueryContext(ctx, sqlStr, vals...)
	if err != nil {
		span.SetStatus(codes.Error, "Query failed")
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	ids := make([]string, 0)
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			span.SetStatus(codes.Error, "Scan row failed")
			return nil, err
		}
		ids = append(ids, id)
	}

	span.SetStatus(codes.Ok, "")
	return ids, nil
}

// List lists Resources with filters.
func (ra *resourceAccess) List(ctx context.Context, params interfaces.ResourcesQueryParams) ([]*interfaces.Resource, int64, error) {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "List resources")
	defer span.End()

	builder := sq.Select(
		resourceExtCol(params, "f_id"),
		resourceExtCol(params, "f_catalog_id"),
		resourceExtCol(params, "f_name"),
		resourceExtCol(params, "f_tags"),
		resourceExtCol(params, "f_description"),
		resourceExtCol(params, "f_category"),
		resourceExtCol(params, "f_status"),
		resourceExtCol(params, "f_status_message"),
		resourceExtCol(params, "f_last_discover_status"),
		resourceExtCol(params, "f_database"),
		resourceExtCol(params, "f_source_identifier"),
		resourceExtCol(params, "f_source_metadata"),
		resourceExtCol(params, "f_schema_definition"),
		resourceExtCol(params, "f_creator"),
		resourceExtCol(params, "f_creator_type"),
		resourceExtCol(params, "f_create_time"),
		resourceExtCol(params, "f_updater"),
		resourceExtCol(params, "f_updater_type"),
		resourceExtCol(params, "f_update_time"),
	).From(RESOURCE_TABLE_NAME)

	countBuilder := sq.Select("COUNT(*)").From(RESOURCE_TABLE_NAME)

	if params.Name != "" {
		name := "%" + common.EscapeLikePattern(params.Name) + "%"
		builder = builder.Where(sq.Like{resourceExtCol(params, "f_name"): name})
		countBuilder = countBuilder.Where(sq.Like{resourceExtCol(params, "f_name"): name})
	}
	if params.CatalogID != "" {
		builder = builder.Where(sq.Eq{resourceExtCol(params, "f_catalog_id"): params.CatalogID})
		countBuilder = countBuilder.Where(sq.Eq{resourceExtCol(params, "f_catalog_id"): params.CatalogID})
	}
	if params.Category != "" {
		builder = builder.Where(sq.Eq{resourceExtCol(params, "f_category"): params.Category})
		countBuilder = countBuilder.Where(sq.Eq{resourceExtCol(params, "f_category"): params.Category})
	}
	if params.Status != "" {
		builder = builder.Where(sq.Eq{resourceExtCol(params, "f_status"): params.Status})
		countBuilder = countBuilder.Where(sq.Eq{resourceExtCol(params, "f_status"): params.Status})
	}
	if params.Database != "" {
		builder = builder.Where(sq.Eq{resourceExtCol(params, "f_database"): params.Database})
		countBuilder = countBuilder.Where(sq.Eq{resourceExtCol(params, "f_database"): params.Database})
	}

	builder = applyResourceExtensionJoins(builder, params)
	countBuilder = applyResourceExtensionJoins(countBuilder, params)

	countSql, countVals, _ := countBuilder.ToSql()
	var total int64
	err := ra.db.QueryRowContext(ctx, countSql, countVals...).Scan(&total)
	if err != nil {
		logger.Errorf("Failed to count resources: %v", err)
		span.SetStatus(codes.Error, "Count failed")
		return nil, 0, err
	}

	// Pagination is applied in service after permission filtering.
	// 排序
	if params.Sort != "" {
		builder = builder.OrderBy(resourceListOrderExpr(params))
	} else {
		if len(params.ExtensionKeys) > 0 {
			builder = builder.OrderBy(fmt.Sprintf("t_resource.f_update_time %s", params.Direction))
		} else {
			builder = builder.OrderBy("f_update_time DESC")
		}
	}

	sqlStr, vals, err := builder.ToSql()
	if err != nil {
		span.SetStatus(codes.Error, "Build sql failed")
		return nil, 0, err
	}

	rows, err := ra.db.QueryContext(ctx, sqlStr, vals...)
	if err != nil {
		span.SetStatus(codes.Error, "Query failed")
		return nil, 0, err
	}
	defer func() { _ = rows.Close() }()

	resources := make([]*interfaces.Resource, 0)
	for rows.Next() {
		resource := &interfaces.Resource{}
		var tagsStr string
		var database, sourceIdentifier, sourceMetadata, schemaDefinition sql.NullString

		err := rows.Scan(
			&resource.ID,
			&resource.CatalogID,
			&resource.Name,
			&tagsStr,
			&resource.Description,
			&resource.Category,
			&resource.Status,
			&resource.StatusMessage,
			&resource.LastDiscoverStatus,
			&database,
			&sourceIdentifier,
			&sourceMetadata,
			&schemaDefinition,
			&resource.Creator.ID,
			&resource.Creator.Type,
			&resource.CreateTime,
			&resource.Updater.ID,
			&resource.Updater.Type,
			&resource.UpdateTime,
		)
		if err != nil {
			span.SetStatus(codes.Error, "Scan row failed")
			return nil, 0, err
		}

		// tags string 转成数组的格式
		resource.Tags = libCommon.TagString2TagSlice(tagsStr)
		resource.Database = database.String
		resource.SourceIdentifier = sourceIdentifier.String
		if sourceMetadata.Valid && sourceMetadata.String != "" {
			_ = sonic.Unmarshal([]byte(sourceMetadata.String), &resource.SourceMetadata)
		}
		if schemaDefinition.Valid && schemaDefinition.String != "" {
			_ = sonic.Unmarshal([]byte(schemaDefinition.String), &resource.SchemaDefinition)
		}

		resources = append(resources, resource)
	}

	if err := attachResourceExtensions(ctx, ra.appSetting, params, resources); err != nil {
		span.SetStatus(codes.Error, "Load resource extensions failed")
		return nil, 0, err
	}

	span.SetStatus(codes.Ok, "")
	return resources, total, nil
}

// Update updates ra Resource.
func (ra *resourceAccess) Update(ctx context.Context, resource *interfaces.Resource) error {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Update resource")
	defer span.End()

	span.SetAttributes(attr.Key("resource_id").String(resource.ID))

	// tags 转成 string 的格式
	tagsStr := libCommon.TagSlice2TagString(resource.Tags)

	// 序列化 SourceMetadata, SchemaDefinition, logicDefinition
	sourceMetadataBytes, _ := sonic.Marshal(resource.SourceMetadata)
	if resource.SourceMetadata == nil {
		sourceMetadataBytes = []byte("{}")
	}
	schemaDefinitionBytes, _ := sonic.Marshal(resource.SchemaDefinition)
	if resource.SchemaDefinition == nil {
		schemaDefinitionBytes = []byte("[]")
	}
	logicDefinitionBytes, _ := sonic.Marshal(resource.LogicDefinition)
	if resource.LogicDefinition == nil {
		logicDefinitionBytes = []byte("[]")
	}

	builder := sq.Update(RESOURCE_TABLE_NAME).
		Set("f_catalog_id", resource.CatalogID).
		Set("f_name", resource.Name).
		Set("f_tags", tagsStr).
		Set("f_description", resource.Description).
		Set("f_source_metadata", string(sourceMetadataBytes)).
		Set("f_schema_definition", string(schemaDefinitionBytes)).
		Set("f_logic_type", resource.LogicType).
		Set("f_logic_definition", string(logicDefinitionBytes)).
		Set("f_updater", resource.Updater.ID).
		Set("f_updater_type", resource.Updater.Type).
		Set("f_update_time", resource.UpdateTime).
		Set("f_local_index_name", resource.LocalIndexName).
		Where(sq.Eq{"f_id": resource.ID})
	if resource.LastDiscoverStatus != "" {
		builder = builder.Set("f_last_discover_status", resource.LastDiscoverStatus)
	}

	sqlStr, vals, err := builder.ToSql()
	if err != nil {
		span.SetStatus(codes.Error, "Build sql failed")
		return err
	}

	_, err = ra.db.ExecContext(ctx, sqlStr, vals...)
	if err != nil {
		span.SetStatus(codes.Error, "Update failed")
		return err
	}

	span.SetStatus(codes.Ok, "")
	return nil
}

// GetByCatalogID retrieves all Resources under a Catalog.
func (ra *resourceAccess) GetByCatalogID(ctx context.Context, catalogID string) ([]*interfaces.Resource, error) {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Query resources by catalog ID")
	defer span.End()

	span.SetAttributes(attr.Key("catalog_id").String(catalogID))

	sqlStr, vals, err := sq.Select(
		"f_id",
		"f_catalog_id",
		"f_name",
		"f_tags",
		"f_description",
		"f_category",
		"f_status",
		"f_status_message",
		"f_last_discover_status",
		"f_database",
		"f_source_identifier",
		"f_source_metadata",
		"f_schema_definition",
		"f_creator",
		"f_creator_type",
		"f_create_time",
		"f_updater",
		"f_updater_type",
		"f_update_time",
	).From(RESOURCE_TABLE_NAME).
		Where(sq.Eq{"f_catalog_id": catalogID}).
		ToSql()
	if err != nil {
		logger.Errorf("Failed to build query resources sql: %v", err)
		span.SetStatus(codes.Error, "Build sql failed")
		return nil, err
	}

	rows, err := ra.db.QueryContext(ctx, sqlStr, vals...)
	if err != nil {
		logger.Errorf("Query resources failed: %v", err)
		span.SetStatus(codes.Error, "Query failed")
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	resources := make([]*interfaces.Resource, 0)
	for rows.Next() {
		resource := &interfaces.Resource{}
		var tagsStr string
		var database, sourceIdentifier, sourceMetadata, schemaDefinition sql.NullString

		err := rows.Scan(
			&resource.ID,
			&resource.CatalogID,
			&resource.Name,
			&tagsStr,
			&resource.Description,
			&resource.Category,
			&resource.Status,
			&resource.StatusMessage,
			&resource.LastDiscoverStatus,
			&database,
			&sourceIdentifier,
			&sourceMetadata,
			&schemaDefinition,
			&resource.Creator.ID,
			&resource.Creator.Type,
			&resource.CreateTime,
			&resource.Updater.ID,
			&resource.Updater.Type,
			&resource.UpdateTime,
		)
		if err != nil {
			logger.Errorf("Scan resource row failed: %v", err)
			span.SetStatus(codes.Error, "Scan row failed")
			return nil, err
		}

		resource.Tags = libCommon.TagString2TagSlice(tagsStr)
		resource.Database = database.String
		resource.SourceIdentifier = sourceIdentifier.String
		if sourceMetadata.Valid && sourceMetadata.String != "" {
			_ = sonic.Unmarshal([]byte(sourceMetadata.String), &resource.SourceMetadata)
		}
		if schemaDefinition.Valid && schemaDefinition.String != "" {
			_ = sonic.Unmarshal([]byte(schemaDefinition.String), &resource.SchemaDefinition)
		}

		resources = append(resources, resource)
	}

	span.SetStatus(codes.Ok, "")
	return resources, nil
}

// UpdateStatus updates a Resource's status.
func (ra *resourceAccess) UpdateStatus(ctx context.Context, id string, status string, statusMessage string) error {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Update resource status")
	defer span.End()

	span.SetAttributes(
		attr.Key("resource_id").String(id),
		attr.Key("status").String(status),
	)

	sqlStr, vals, err := sq.Update(RESOURCE_TABLE_NAME).
		Set("f_status", status).
		Set("f_status_message", statusMessage).
		Where(sq.Eq{"f_id": id}).
		ToSql()
	if err != nil {
		span.SetStatus(codes.Error, "Build sql failed")
		return err
	}

	_, err = ra.db.ExecContext(ctx, sqlStr, vals...)
	if err != nil {
		span.SetStatus(codes.Error, "Update failed")
		return err
	}

	span.SetStatus(codes.Ok, "")
	return nil
}

// UpdateDiscoverStatus updates a Resource's last discover status.
func (ra *resourceAccess) UpdateDiscoverStatus(ctx context.Context, id string, status string) error {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Update resource discover status")
	defer span.End()

	span.SetAttributes(
		attr.Key("resource_id").String(id),
		attr.Key("last_discover_status").String(status),
	)

	sqlStr, vals, err := sq.Update(RESOURCE_TABLE_NAME).
		Set("f_last_discover_status", status).
		Where(sq.Eq{"f_id": id}).
		ToSql()
	if err != nil {
		span.SetStatus(codes.Error, "Build sql failed")
		return err
	}

	_, err = ra.db.ExecContext(ctx, sqlStr, vals...)
	if err != nil {
		span.SetStatus(codes.Error, "Update failed")
		return err
	}

	span.SetStatus(codes.Ok, "")
	return nil
}

func (ra *resourceAccess) DeleteByIDs(ctx context.Context, ids []string) error {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Delete resources")
	defer span.End()

	span.SetAttributes(attr.Key("resource_ids").StringSlice(ids))

	if len(ids) == 0 {
		return nil
	}

	if err := entityextension.NewStore(ra.appSetting).DeleteByEntityIDs(ctx, entityextension.KindResource, ids); err != nil {
		span.SetStatus(codes.Error, "Delete entity extensions failed")
		return err
	}

	sqlStr, vals, _ := sq.Delete(RESOURCE_TABLE_NAME).
		Where(sq.Eq{"f_id": ids}).
		ToSql()

	_, err := ra.db.ExecContext(ctx, sqlStr, vals...)
	if err != nil {
		span.SetStatus(codes.Error, "Delete failed")
		return err
	}

	span.SetStatus(codes.Ok, "")
	return nil
}

// ListAuthResources lists resource auth resources with filters.
func (ra *resourceAccess) ListAuthResources(ctx context.Context, params interfaces.AuthResourceQueryParams) ([]*interfaces.AuthResourceEntry, error) {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "ListAuthResources")
	defer span.End()

	builder := sq.Select(
		"f_id",
		"f_name",
	).From(RESOURCE_TABLE_NAME)

	if params.ID != "" {
		builder = builder.Where(sq.Eq{"f_id": params.ID})
	}

	if params.Keyword != "" {
		keyword := "%" + common.EscapeLikePattern(params.Keyword) + "%"
		builder = builder.Where(sq.Like{"f_name": keyword})
	}

	// 排序
	if params.Sort != "" {
		builder = builder.OrderBy(fmt.Sprintf("%s %s", params.Sort, params.Direction))
	} else {
		builder = builder.OrderBy("f_update_time DESC")
	}

	sqlStr, vals, err := builder.ToSql()
	if err != nil {
		span.SetStatus(codes.Error, "Build sql failed")
		return nil, err
	}

	rows, err := ra.db.QueryContext(ctx, sqlStr, vals...)
	if err != nil {
		span.SetStatus(codes.Error, "Query failed")
		return nil, err
	}
	defer func() { _ = rows.Close() }()

	entries := make([]*interfaces.AuthResourceEntry, 0)
	for rows.Next() {
		entry := &interfaces.AuthResourceEntry{}

		err := rows.Scan(
			&entry.ID,
			&entry.Name,
		)
		if err != nil {
			span.SetStatus(codes.Error, "Scan row failed")
			return nil, err
		}
		entry.Type = interfaces.AUTH_RESOURCE_TYPE_RESOURCE
		entries = append(entries, entry)
	}

	span.SetStatus(codes.Ok, "")
	return entries, nil
}

func (ra *resourceAccess) CheckExistByCategories(ctx context.Context, catalogID string, categories []string) (bool, error) {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Check resources exist")
	defer span.End()

	span.SetAttributes(attr.Key("catalog_id").String(catalogID))

	countBuilder := sq.Select("COUNT(*)").From(RESOURCE_TABLE_NAME)

	if catalogID != "" {
		countBuilder = countBuilder.Where(sq.Eq{"f_catalog_id": catalogID})
	}
	if len(categories) > 0 {
		countBuilder = countBuilder.Where(sq.Eq{"f_category": categories})
	}

	countSql, countVals, _ := countBuilder.ToSql()
	var total int64
	err := ra.db.QueryRowContext(ctx, countSql, countVals...).Scan(&total)
	if err != nil {
		logger.Errorf("Failed to count resources: %v", err)
		span.SetStatus(codes.Error, "Count failed")
		return false, err
	}

	span.SetStatus(codes.Ok, "")
	return total > 0, nil
}

func (ra *resourceAccess) DeleteByCatalogIDs(ctx context.Context, catalogIDs []string) error {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Delete resources by catalog IDs")
	defer span.End()

	span.SetAttributes(attr.Key("catalog_ids").StringSlice(catalogIDs))

	if len(catalogIDs) == 0 {
		return nil
	}

	idSql, idArgs, err := sq.Select("f_id").From(RESOURCE_TABLE_NAME).Where(sq.Eq{"f_catalog_id": catalogIDs}).ToSql()
	if err != nil {
		span.SetStatus(codes.Error, "Build sql failed")
		return err
	}
	ridRows, err := ra.db.QueryContext(ctx, idSql, idArgs...)
	if err != nil {
		span.SetStatus(codes.Error, "Query resource ids failed")
		return err
	}
	defer func() { _ = ridRows.Close() }()

	var resIDs []string
	for ridRows.Next() {
		var rid string
		if err := ridRows.Scan(&rid); err != nil {
			span.SetStatus(codes.Error, "Scan failed")
			return err
		}
		resIDs = append(resIDs, rid)
	}
	if err := ridRows.Err(); err != nil {
		span.SetStatus(codes.Error, "Iterate rows failed")
		return err
	}
	if err := entityextension.NewStore(ra.appSetting).DeleteByEntityIDs(ctx, entityextension.KindResource, resIDs); err != nil {
		span.SetStatus(codes.Error, "Delete entity extensions failed")
		return err
	}

	sqlStr, vals, _ := sq.Delete(RESOURCE_TABLE_NAME).
		Where(sq.Eq{"f_catalog_id": catalogIDs}).
		ToSql()

	_, execErr := ra.db.ExecContext(ctx, sqlStr, vals...)
	if execErr != nil {
		span.SetStatus(codes.Error, "Delete failed")
		return execErr
	}

	span.SetStatus(codes.Ok, "")
	return nil
}
