package records

import (
	"context"
	"fmt"
	"sort"
	"strings"

	appErrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/recordlocks"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/models"
)

func (ps *PersistenceService) filterExclusiveSiblingBulkInsertLockable(ctx context.Context, schema *models.ObjectMetadata, records []models.SObject, indexMap []int, rejectionLog *bulkRejectionLog) ([]models.SObject, []int, error) {
	contracts, err := ps.exclusiveSiblingContracts(schema)
	if err != nil || len(contracts) == 0 || len(records) == 0 {
		return records, indexMap, err
	}

	keyRecords := map[string][]int{}
	keysByID := map[string]recordlocks.ParentKey{}
	for i, record := range records {
		for _, contract := range contracts {
			if !exclusiveSiblingTruthy(record[contract.FieldAPIName]) {
				continue
			}
			partition := exclusiveSiblingRecordID(record[contract.PartitionFieldAPIName])
			if partition == "" {
				continue
			}
			key := recordlocks.ParentKey{ObjectName: contract.ParentObjectAPIName, RecordID: partition}
			keyID := recordlocks.KeyID(key)
			if keyID == "\x00" {
				continue
			}
			keysByID[keyID] = key
			keyRecords[keyID] = append(keyRecords[keyID], i)
		}
	}
	if len(keysByID) == 0 {
		return records, indexMap, nil
	}

	keys := make([]recordlocks.ParentKey, 0, len(keysByID))
	for _, key := range keysByID {
		keys = append(keys, key)
	}
	failed := map[int]struct{}{}
	for _, key := range recordlocks.OrderedParentKeys(keys) {
		if _, err := ps.LockParentRows(ctx, []recordlocks.ParentKey{key}); err != nil {
			if !appErrors.IsNotFound(err) {
				return nil, nil, err
			}
			for _, recordIndex := range keyRecords[recordlocks.KeyID(key)] {
				if _, alreadyFailed := failed[recordIndex]; alreadyFailed {
					continue
				}
				failed[recordIndex] = struct{}{}
				originalIndex := recordIndex
				if recordIndex < len(indexMap) {
					originalIndex = indexMap[recordIndex]
				}
				appendBulkInsertRecordError(rejectionLog, originalIndex, bulkRecordAt(records, recordIndex), appErrors.NewValidationError("ExclusiveSiblingFields", fmt.Sprintf("parent record %s/%s not found", key.ObjectName, key.RecordID)))
			}
		}
	}
	if len(failed) == 0 {
		return records, indexMap, nil
	}

	filteredRecords := make([]models.SObject, 0, len(records)-len(failed))
	filteredIndexMap := make([]int, 0, len(indexMap))
	for i, record := range records {
		if _, ok := failed[i]; ok {
			continue
		}
		filteredRecords = append(filteredRecords, record)
		if i < len(indexMap) {
			filteredIndexMap = append(filteredIndexMap, indexMap[i])
		}
	}
	return filteredRecords, filteredIndexMap, nil
}

func (ps *PersistenceService) exclusiveSiblingBulkInsertResetPlans(schema *models.ObjectMetadata, records []models.SObject, landedRows []bool) ([]exclusiveSiblingResetPlan, error) {
	contracts, err := ps.exclusiveSiblingContracts(schema)
	if err != nil || len(contracts) == 0 {
		return nil, err
	}
	type partitionState struct {
		contract       exclusiveSiblingContract
		partitionValue string
		landedTrueIDs  []string
	}
	states := map[string]*partitionState{}
	for i, record := range records {
		if i >= len(landedRows) || !landedRows[i] {
			continue
		}
		recordID := exclusiveSiblingRecordID(record[constants.FieldID])
		if recordID == "" {
			continue
		}
		for _, contract := range contracts {
			if !exclusiveSiblingTruthy(record[contract.FieldAPIName]) {
				continue
			}
			partition := exclusiveSiblingRecordID(record[contract.PartitionFieldAPIName])
			if partition == "" {
				continue
			}
			key := exclusiveSiblingBulkInsertPartitionKey(contract, partition)
			state := states[key]
			if state == nil {
				state = &partitionState{contract: contract, partitionValue: partition}
				states[key] = state
			}
			state.landedTrueIDs = append(state.landedTrueIDs, recordID)
		}
	}
	if len(states) == 0 {
		return nil, nil
	}

	keys := make([]string, 0, len(states))
	for key := range states {
		keys = append(keys, key)
	}
	sort.Strings(keys)
	plans := make([]exclusiveSiblingResetPlan, 0, len(keys))
	for _, key := range keys {
		state := states[key]
		if state == nil || len(state.landedTrueIDs) == 0 {
			continue
		}
		winnerID := state.landedTrueIDs[len(state.landedTrueIDs)-1]
		skipped := map[string]struct{}{}
		for _, id := range state.landedTrueIDs[:len(state.landedTrueIDs)-1] {
			if id != "" && id != winnerID {
				skipped[id] = struct{}{}
			}
		}
		plans = append(plans, exclusiveSiblingResetPlan{
			contract:          state.contract,
			partitionValue:    state.partitionValue,
			winnerID:          winnerID,
			skippedSiblingIDs: skipped,
		})
	}
	return plans, nil
}

func exclusiveSiblingBulkInsertPartitionKey(contract exclusiveSiblingContract, partitionValue string) string {
	return strings.ToLower(contract.FieldAPIName) + "\x00" + strings.ToLower(contract.PartitionFieldAPIName) + "\x00" + partitionValue
}

func (ps *PersistenceService) applyExclusiveSiblingBulkInsertLandedRowResets(ctx context.Context, schema *models.ObjectMetadata, records []models.SObject, landedRows []bool, plans []exclusiveSiblingResetPlan) (map[string]models.SObject, error) {
	updatesByID := exclusiveSiblingBulkInsertLandedRowResetUpdates(plans)
	finalByID, err := ps.previewExclusiveSiblingBulkInsertFinalRecords(ctx, schema, records, landedRows, updatesByID)
	if err != nil {
		return nil, err
	}

	ids := make([]string, 0, len(finalByID))
	for id := range finalByID {
		ids = append(ids, id)
	}
	sort.Strings(ids)
	recordsByID := recordsByIDForLandedRows(records, landedRows)
	for _, id := range ids {
		finalRecord := finalByID[id]
		logicalUpdates := exclusiveSiblingChangedLogicalUpdates(schema, recordsByID[id], finalRecord)
		if len(logicalUpdates) > 0 {
			if err := ps.repo.Update(ctx, schema.APIName, id, logicalUpdates); err != nil {
				return nil, appErrors.NewInternalError("exclusive sibling bulk landed-row reset failed", err)
			}
			InvalidateRecordReadCache(ctx, schema.APIName, id)
		}
	}
	return finalByID, nil
}

func exclusiveSiblingBulkInsertLandedRowResetUpdates(plans []exclusiveSiblingResetPlan) map[string]models.SObject {
	updatesByID := map[string]models.SObject{}
	for _, plan := range plans {
		for id := range plan.skippedSiblingIDs {
			if updatesByID[id] == nil {
				updatesByID[id] = models.SObject{}
			}
			updatesByID[id][plan.contract.FieldAPIName] = false
		}
	}
	return updatesByID
}

func recordsByIDForLandedRows(records []models.SObject, landedRows []bool) map[string]models.SObject {
	recordsByID := map[string]models.SObject{}
	for i, record := range records {
		if i >= len(landedRows) || !landedRows[i] {
			continue
		}
		if id := exclusiveSiblingRecordID(record[constants.FieldID]); id != "" {
			recordsByID[id] = record
		}
	}
	return recordsByID
}

func (ps *PersistenceService) previewExclusiveSiblingBulkInsertFinalRecords(ctx context.Context, schema *models.ObjectMetadata, records []models.SObject, landedRows []bool, updatesByID map[string]models.SObject) (map[string]models.SObject, error) {
	if len(updatesByID) == 0 {
		return nil, nil
	}
	recordsByID := recordsByIDForLandedRows(records, landedRows)

	ids := make([]string, 0, len(updatesByID))
	for id := range updatesByID {
		ids = append(ids, id)
	}
	sort.Strings(ids)
	finalByID := map[string]models.SObject{}
	for _, id := range ids {
		oldRecord := recordsByID[id]
		if oldRecord == nil {
			continue
		}
		finalRecord, logicalUpdates, err := ps.exclusiveSiblingResetFinalUpdates(ctx, schema, oldRecord, updatesByID[id])
		if err != nil {
			return nil, err
		}
		if len(logicalUpdates) == 0 {
			continue
		}
		finalByID[id] = finalRecord
	}
	return finalByID, nil
}

func applyExclusiveSiblingBulkInsertFinalRecords(records []models.SObject, landedRows []bool, finalByID map[string]models.SObject) {
	if len(records) == 0 || len(finalByID) == 0 {
		return
	}
	for i, record := range records {
		if i >= len(landedRows) || !landedRows[i] {
			continue
		}
		id := exclusiveSiblingRecordID(record[constants.FieldID])
		for fieldName, value := range finalByID[id] {
			record[fieldName] = value
		}
	}
}