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

// Package connectortype provides ConnectorType data access operations.
package connector_type

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/interfaces"
)

const (
	CONNECTOR_TYPE_TABLE_NAME = "t_connector_type"
)

var (
	ctAccessOnce sync.Once
	ctAccess     interfaces.ConnectorTypeAccess
)

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

type connectorTypeScanner interface {
	Scan(dest ...any) error
}

func connectorTypeColumns() []string {
	return []string{
		"f_type",
		"f_name",
		"f_tags",
		"f_description",
		"f_mode",
		"f_category",
		"f_endpoint",
		"f_field_config",
		"f_enabled",
	}
}

func scanConnectorType(scanner connectorTypeScanner) (*interfaces.ConnectorType, error) {
	ct := &interfaces.ConnectorType{}
	var tagsStr string
	var fieldConfigStr string

	err := scanner.Scan(
		&ct.Type,
		&ct.Name,
		&tagsStr,
		&ct.Description,
		&ct.Mode,
		&ct.Category,
		&ct.Endpoint,
		&fieldConfigStr,
		&ct.Enabled,
	)
	if err != nil {
		return nil, err
	}

	ct.Tags = libCommon.TagString2TagSlice(tagsStr)

	if fieldConfigStr != "" {
		if err := sonic.UnmarshalString(fieldConfigStr, &ct.FieldConfig); err != nil {
			return nil, err
		}
	}

	return ct, nil
}

// NewConnectorTypeAccess creates a new ConnectorTypeAccess.
func NewConnectorTypeAccess(appSetting *common.AppSetting) interfaces.ConnectorTypeAccess {
	ctAccessOnce.Do(func() {
		ctAccess = &connectorTypeAccess{
			appSetting: appSetting,
			db:         libdb.NewDB(&appSetting.DBSetting),
		}
	})
	return ctAccess
}

// Create creates a new ConnectorType.
func (cta *connectorTypeAccess) Create(ctx context.Context, ct *interfaces.ConnectorType) error {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Insert into connector_type")
	defer span.End()

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

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

	// Serialize FieldConfig to JSON
	fieldConfigStr, err := sonic.MarshalString(ct.FieldConfig)
	if err != nil {
		otellog.LogError(ctx, "Failed to marshal field config", err)
		return err
	}

	sqlStr, vals, err := sq.Insert(CONNECTOR_TYPE_TABLE_NAME).
		Columns(connectorTypeColumns()...).
		Values(
			ct.Type,
			ct.Name,
			tagsStr,
			ct.Description,
			string(ct.Mode),
			string(ct.Category),
			ct.Endpoint,
			fieldConfigStr,
			ct.Enabled,
		).ToSql()
	if err != nil {
		otellog.LogError(ctx, "Failed to build insert connector_type sql", err)
		return err
	}

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

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

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

// GetByType retrieves a ConnectorType by Type.
func (cta *connectorTypeAccess) GetByType(ctx context.Context, tp string) (*interfaces.ConnectorType, error) {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Query connector_type by Type")
	defer span.End()

	span.SetAttributes(attr.Key("connector_type").String(tp))

	sqlStr, vals, err := sq.Select(connectorTypeColumns()...).
		From(CONNECTOR_TYPE_TABLE_NAME).
		Where(sq.Eq{"f_type": tp}).
		ToSql()
	if err != nil {
		logger.Errorf("Failed to build select connector_type sql: %v", err)
		span.SetStatus(codes.Error, "Build sql failed")
		return nil, err
	}

	row := cta.db.QueryRowContext(ctx, sqlStr, vals...)
	ct, err := scanConnectorType(row)
	if err == sql.ErrNoRows {
		span.SetStatus(codes.Ok, "")
		return nil, nil
	}
	if err != nil {
		logger.Errorf("Scan connector_type failed: %v", err)
		span.SetStatus(codes.Error, "Scan failed")
		return nil, err
	}

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

// GetByName retrieves a ConnectorType by Name.
func (cta *connectorTypeAccess) GetByName(ctx context.Context, name string) (*interfaces.ConnectorType, error) {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Query connector_type by Name")
	defer span.End()

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

	sqlStr, vals, err := sq.Select(connectorTypeColumns()...).
		From(CONNECTOR_TYPE_TABLE_NAME).
		Where(sq.Eq{"f_name": name}).
		ToSql()
	if err != nil {
		logger.Errorf("Failed to build select connector_type sql: %v", err)
		span.SetStatus(codes.Error, "Build sql failed")
		return nil, err
	}

	row := cta.db.QueryRowContext(ctx, sqlStr, vals...)
	ct, err := scanConnectorType(row)
	if err == sql.ErrNoRows {
		span.SetStatus(codes.Ok, "")
		return nil, nil
	}
	if err != nil {
		logger.Errorf("Scan connector_type failed: %v", err)
		span.SetStatus(codes.Error, "Scan failed")
		return nil, err
	}

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

// List lists ConnectorTypes with filters.
func (cta *connectorTypeAccess) List(ctx context.Context, params interfaces.ConnectorTypesQueryParams) ([]*interfaces.ConnectorType, int64, error) {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "List connector_types")
	defer span.End()

	builder := sq.Select(connectorTypeColumns()...).
		From(CONNECTOR_TYPE_TABLE_NAME)

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

	// Apply filters
	if params.Name != "" {
		name := "%" + common.EscapeLikePattern(params.Name) + "%"
		builder = builder.Where(sq.Like{"f_name": name})
		countBuilder = countBuilder.Where(sq.Like{"f_name": name})
	}
	if params.Mode != "" {
		builder = builder.Where(sq.Eq{"f_mode": params.Mode})
		countBuilder = countBuilder.Where(sq.Eq{"f_mode": params.Mode})
	}
	if params.Category != "" {
		builder = builder.Where(sq.Eq{"f_category": params.Category})
		countBuilder = countBuilder.Where(sq.Eq{"f_category": params.Category})
	}
	if params.Enabled != nil {
		builder = builder.Where(sq.Eq{"f_enabled": *params.Enabled})
		countBuilder = countBuilder.Where(sq.Eq{"f_enabled": *params.Enabled})
	}

	countSql, countVals, _ := countBuilder.ToSql()
	var total int64
	err := cta.db.QueryRowContext(ctx, countSql, countVals...).Scan(&total)
	if err != nil {
		logger.Errorf("Count connector_type failed: %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(fmt.Sprintf("%s %s", params.Sort, params.Direction))
	} else {
		builder = builder.OrderBy("f_name ASC")
	}

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

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

	connectorTypes := make([]*interfaces.ConnectorType, 0)
	for rows.Next() {
		ct, err := scanConnectorType(rows)
		if err != nil {
			span.SetStatus(codes.Error, "Scan row failed")
			return nil, 0, err
		}

		connectorTypes = append(connectorTypes, ct)
	}

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

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

	builder := sq.Select(
		"f_type",
		"f_name",
	).From(CONNECTOR_TYPE_TABLE_NAME)

	if params.ID != "" {
		builder = builder.Where(sq.Eq{"f_type": 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_name ASC")
	}

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

	rows, err := cta.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{}
		if err := rows.Scan(&entry.ID, &entry.Name); err != nil {
			span.SetStatus(codes.Error, "Scan row failed")
			return nil, err
		}
		entry.Type = interfaces.AuthResourceTypeConnectorType
		entries = append(entries, entry)
	}

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

// Update updates a ConnectorType.
func (cta *connectorTypeAccess) Update(ctx context.Context, ct *interfaces.ConnectorType) error {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Update connector_type")
	defer span.End()

	span.SetAttributes(attr.Key("connector_type").String(ct.Type))

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

	// Serialize FieldConfig to JSON
	fieldConfigStr, err := sonic.MarshalString(ct.FieldConfig)
	if err != nil {
		span.SetStatus(codes.Error, "Marshal field config failed")
		return err
	}

	sqlStr, vals, err := sq.Update(CONNECTOR_TYPE_TABLE_NAME).
		Set("f_name", ct.Name).
		Set("f_tags", tagsStr).
		Set("f_description", ct.Description).
		Set("f_mode", string(ct.Mode)).
		Set("f_category", string(ct.Category)).
		Set("f_endpoint", ct.Endpoint).
		Set("f_field_config", fieldConfigStr).
		Set("f_enabled", ct.Enabled).
		Where(sq.Eq{"f_type": ct.Type}).
		ToSql()
	if err != nil {
		span.SetStatus(codes.Error, "Build sql failed")
		return err
	}

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

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

// Delete deletes a ConnectorType by Type.
func (cta *connectorTypeAccess) DeleteByType(ctx context.Context, tp string) error {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Delete connector_type")
	defer span.End()

	span.SetAttributes(attr.Key("connector_type").String(tp))

	sqlStr, vals, _ := sq.Delete(CONNECTOR_TYPE_TABLE_NAME).
		Where(sq.Eq{"f_type": tp}).
		ToSql()

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

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

// SetEnabled enables/disables a ConnectorType.
func (cta *connectorTypeAccess) SetEnabled(ctx context.Context, tp string, enabled bool) error {
	ctx, span := oteltrace.StartNamedClientSpan(ctx, "Set connector_type enabled")
	defer span.End()

	span.SetAttributes(attr.Key("connector_type").String(tp))

	sqlStr, vals, _ := sq.Update(CONNECTOR_TYPE_TABLE_NAME).
		Set("f_enabled", enabled).
		Where(sq.Eq{"f_type": tp}).
		ToSql()

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

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