package persistence

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"sort"
	"strings"

	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

// RecordAttachmentSourceIdentifierSegment is the Catalog-declared relation
// between a product-parsed source value and a bounded byte segment of the
// source object's physical record identifier. Nil means the source value is
// the complete source record identifier. Persistence does not interpret URL
// grammar or product record-ID prefixes.
type RecordAttachmentSourceIdentifierSegment struct {
	StartOffsetBytes uint16
	LengthBytes      uint16
}

// RecordAttachmentSourcePlan is the already Catalog-validated physical lookup
// closure. Persistence owns SQL construction; it does not own resolver
// identity, parent-field proof, or authorization.
type RecordAttachmentSourcePlan struct {
	SourceObject       *models.ObjectMetadata
	FileReferenceField string
	DeletedField       string
	IdentifierSegment  *RecordAttachmentSourceIdentifierSegment
}

type RecordAttachmentSourceRepository struct{ db *sql.DB }

func NewRecordAttachmentSourceRepository(db *sql.DB) *RecordAttachmentSourceRepository {
	return &RecordAttachmentSourceRepository{db: db}
}

// Lookup returns the newest active protected file ID for each proof-bound
// source value under one exact parent object and record identity. One batch
// produces one SQL statement, so a bounded continuation cannot become a
// source-value N+1. Values are never normalized: the records owner has proved
// their exact bytes in the current authorized parent field. Every equality and
// identifier segment below is evaluated as a binary string; the Catalog and
// Runtime proof contract is byte-based, not the database's default collation.
func (r *RecordAttachmentSourceRepository) Lookup(
	ctx context.Context,
	plan RecordAttachmentSourcePlan,
	parentObjectID string,
	parentID string,
	sourceValues []string,
) (map[string]string, error) {
	if r == nil || r.db == nil || plan.SourceObject == nil {
		return nil, errors.New("record attachment source repository is not configured")
	}
	parentObjectID = strings.TrimSpace(parentObjectID)
	parentID = strings.TrimSpace(parentID)
	if parentObjectID == "" || parentID == "" || len(sourceValues) == 0 {
		return map[string]string{}, nil
	}

	idField := models.FindField(plan.SourceObject, constants.FieldID)
	fileField := models.FindField(plan.SourceObject, plan.FileReferenceField)
	if idField == nil || fileField == nil || fileField.IsVirtual || fileField.IsCompound {
		return nil, errors.New("record attachment source plan references missing physical fields")
	}
	deletedField := (*models.FieldMetadata)(nil)
	if strings.TrimSpace(plan.DeletedField) != "" {
		deletedField = models.FindField(plan.SourceObject, plan.DeletedField)
		if deletedField == nil || deletedField.IsVirtual || deletedField.IsCompound {
			return nil, errors.New("record attachment source plan references a missing physical deleted field")
		}
	}

	table, err := plan.SourceObject.ResolveReadSourceTable()
	if err != nil {
		return nil, fmt.Errorf("resolve record attachment source read source: %w", err)
	}
	table = strings.TrimSpace(table)
	idColumn := resolvePhysicalName(plan.SourceObject, idField.APIName)
	fileColumn := resolvePhysicalName(plan.SourceObject, fileField.APIName)
	deletedColumn := ""
	if deletedField != nil {
		deletedColumn = resolvePhysicalName(plan.SourceObject, deletedField.APIName)
	}
	for _, identifier := range []string{table, idColumn, fileColumn} {
		if !isValidFieldName(identifier) || strings.Contains(identifier, ".") {
			return nil, fmt.Errorf("unsafe record attachment source identifier %q", identifier)
		}
	}
	if deletedColumn != "" && (!isValidFieldName(deletedColumn) || strings.Contains(deletedColumn, ".")) {
		return nil, fmt.Errorf("unsafe record attachment source identifier %q", deletedColumn)
	}

	unique := make([]string, 0, len(sourceValues))
	seen := map[string]struct{}{}
	for _, value := range sourceValues {
		if value == "" {
			continue
		}
		if _, exists := seen[value]; exists {
			continue
		}
		seen[value] = struct{}{}
		unique = append(unique, value)
	}
	if len(unique) == 0 {
		return map[string]string{}, nil
	}
	sort.Strings(unique)

	sourceExpression, err := recordAttachmentSourceIdentifierExpression(idColumn, plan.IdentifierSegment)
	if err != nil {
		return nil, err
	}
	placeholders := strings.TrimSuffix(strings.Repeat("?,", len(unique)), ",")
	query := fmt.Sprintf(
		"SELECT %s, sf.`%s` FROM `%s` src JOIN `%s` sf ON BINARY src.`%s` = BINARY sf.`%s` WHERE %s IN (%s) AND BINARY sf.`%s` = ? AND BINARY sf.`%s` = ? AND (sf.`%s` IS NULL OR sf.`%s` = 'Active')",
		sourceExpression, constants.FieldID, table, constants.TableFile, fileColumn,
		constants.FieldSysFile_SourceContentVersionID, sourceExpression, placeholders,
		constants.FieldSysFile_ParentID, constants.FieldSysFile_ParentObjectAPIName,
		constants.FieldSysFile_Status, constants.FieldSysFile_Status,
	)
	if deletedColumn != "" {
		query += fmt.Sprintf(" AND (src.`%s` IS NULL OR src.`%s` = 0)", deletedColumn, deletedColumn)
	}
	query += fmt.Sprintf(" ORDER BY %s, sf.`%s` DESC", sourceExpression, constants.FieldSysFile_CreatedDate)

	args := make([]any, 0, len(unique)+2)
	for _, value := range unique {
		args = append(args, value)
	}
	args = append(args, parentID, parentObjectID)
	rows, err := r.db.QueryContext(ctx, query, args...)
	if err != nil {
		return nil, fmt.Errorf("lookup record attachment sources: %w", err)
	}
	defer rows.Close()
	result := make(map[string]string, len(unique))
	for rows.Next() {
		var sourceValue, fileID string
		if err := rows.Scan(&sourceValue, &fileID); err != nil {
			return nil, fmt.Errorf("scan record attachment source: %w", err)
		}
		if sourceValue != "" && strings.TrimSpace(fileID) != "" {
			if _, exists := result[sourceValue]; !exists {
				result[sourceValue] = strings.TrimSpace(fileID)
			}
		}
	}
	if err := rows.Err(); err != nil {
		return nil, fmt.Errorf("iterate record attachment sources: %w", err)
	}
	return result, nil
}

func recordAttachmentSourceIdentifierExpression(idColumn string, segment *RecordAttachmentSourceIdentifierSegment) (string, error) {
	// CAST(... AS BINARY) has two required effects in TiDB/MySQL: comparison
	// with a bound source value is exact (rather than collation-equivalent),
	// and SUBSTRING counts bytes (rather than UTF-8 characters). Do not replace
	// it with a column collation: collation alone cannot give byte offsets to
	// SUBSTRING.
	identifier := fmt.Sprintf("CAST(src.`%s` AS BINARY)", idColumn)
	if segment == nil {
		return identifier, nil
	}
	if segment.LengthBytes == 0 || int(segment.StartOffsetBytes)+int(segment.LengthBytes) > 255 {
		return "", errors.New("record attachment source plan has an invalid identifier segment")
	}
	// MySQL/TiDB SUBSTRING positions are one-based. Because identifier is a
	// binary string, positions and length are bytes. The values are immutable
	// Catalog facts, not caller-controlled SQL fragments.
	return fmt.Sprintf("SUBSTRING(%s, %d, %d)", identifier, int(segment.StartOffsetBytes)+1, segment.LengthBytes), nil
}