package analytics

import (
	"context"
	"testing"

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

// ---------------------------------------------------------------------------
// COUNT update-delta path tests (processCountUpdateRollupsIncrementally)
//
// These verify that a child UPDATE applies ±1 COUNT deltas to the affected
// parents instead of re-COUNTing every sibling, and that the path fails closed
// to full recompute when correctness can't be guaranteed.
// ---------------------------------------------------------------------------

// seqMatchRepo returns a different filter-match answer on the first call
// (the pre-update snapshot probe) than on subsequent calls (the live newMatch
// probe), so a single filter can model a child row whose match flips across the
// update. It also counts recompute calls so tests can assert the delta path was
// taken (no recompute) vs. the fallback (recompute happened).
type seqMatchRepo struct {
	stubRollupCalcRepo
	matchSeq  []bool
	matchCall int
}

func (r *seqMatchRepo) RecordMatchesRollupFilter(_ context.Context, _, _ string, _ *string) (bool, error) {
	idx := r.matchCall
	r.matchCall++
	if idx < len(r.matchSeq) {
		return r.matchSeq[idx], nil
	}
	if len(r.matchSeq) == 0 {
		return false, nil
	}
	return r.matchSeq[len(r.matchSeq)-1], nil
}

type batchMatchCall struct {
	summaryObject string
	recordID      string
	filters       []string
}

type batchMatchRepo struct {
	stubRollupCalcRepo
	results     [][]bool
	batchCalls  []batchMatchCall
	singleCalls int
}

func (r *batchMatchRepo) RecordMatchesRollupFilter(_ context.Context, _, _ string, _ *string) (bool, error) {
	r.singleCalls++
	return false, nil
}

func (r *batchMatchRepo) RecordMatchesRollupFilters(
	_ context.Context,
	summaryObject string,
	recordID string,
	filters []*string,
) ([]bool, error) {
	call := batchMatchCall{summaryObject: summaryObject, recordID: recordID}
	for _, filter := range filters {
		call.filters = append(call.filters, rollupFilterKey(filter))
	}
	r.batchCalls = append(r.batchCalls, call)
	index := len(r.batchCalls) - 1
	if index >= len(r.results) {
		return make([]bool, len(filters)), nil
	}
	return append([]bool(nil), r.results[index]...), nil
}

func countSchema(parentObj, childObj, rollupField, relField string, filter *string) map[string]*models.ObjectMetadata {
	return map[string]*models.ObjectMetadata{
		childObj: {APIName: childObj},
		parentObj: {
			APIName: parentObj,
			Fields: []models.FieldMetadata{
				{
					APIName: rollupField,
					Type:    constants.FieldTypeRollupSummary,
					RollupConfig: &models.RollupConfig{
						CalcType:          "COUNT",
						SummaryObject:     childObj,
						SummaryField:      "Id",
						RelationshipField: relField,
						Filter:            filter,
					},
				},
			},
		},
	}
}

func TestCountUpdateDeltaBatchesAllFilterReadsAtEachImmutableImage(t *testing.T) {
	const (
		parentObj = "Campaign"
		childObj  = "CampaignMember"
		relField  = "CampaignId"
		parentID  = "campaign-1"
		childID   = "member-1"
	)
	filters := []string{
		"Status = 'Sent'",
		"Status = 'Responded'",
		"Type = 'Contact'",
		"Type = 'Lead'",
	}
	fields := make([]models.FieldMetadata, 0, len(filters))
	parentValues := models.SObject{}
	for index := range filters {
		fieldAPI := "Count" + string(rune('A'+index))
		filter := filters[index]
		fields = append(fields, models.FieldMetadata{
			APIName: fieldAPI,
			Type:    constants.FieldTypeRollupSummary,
			RollupConfig: &models.RollupConfig{
				CalcType: "COUNT", SummaryObject: childObj, SummaryField: "Id",
				RelationshipField: relField, Filter: &filter,
			},
		})
		parentValues[fieldAPI] = float64(10)
	}
	meta := newFakeMetadata(map[string]*models.ObjectMetadata{
		childObj:  {APIName: childObj},
		parentObj: {APIName: parentObj, Fields: fields},
	})
	repo := &batchMatchRepo{results: [][]bool{
		{false, false, false, false},
		{true, true, true, true},
	}}
	svc := NewRollupService(repo, meta, alwaysPessimisticRollupTxManager{}, noopLogger())
	dw := &stubDerivedWriteCapture{rows: map[string]models.SObject{
		parentObj + "|" + parentID: parentValues,
	}}
	svc.SetDerivedWriter(dw)

	oldRecord := models.SObject{constants.FieldID: childID, relField: parentID}
	finalRecord := models.SObject{constants.FieldID: childID, relField: parentID}
	ctx, err := svc.StageRollupCountUpdatePreimage(context.Background(), childObj, oldRecord)
	if err != nil {
		t.Fatalf("StageRollupCountUpdatePreimage error: %v", err)
	}
	if err := svc.ProcessRollupsForUpdate(ctx, childObj, finalRecord, oldRecord); err != nil {
		t.Fatalf("ProcessRollupsForUpdate error: %v", err)
	}

	if repo.singleCalls != 0 {
		t.Fatalf("single filter reads = %d, want 0", repo.singleCalls)
	}
	if len(repo.batchCalls) != 2 {
		t.Fatalf("batch filter reads = %d, want exactly 2 (pre and post image)", len(repo.batchCalls))
	}
	for index, call := range repo.batchCalls {
		if call.summaryObject != childObj || call.recordID != childID || len(call.filters) != 4 {
			t.Fatalf("batch call %d = %#v, want CampaignMember/member-1 with four filters", index, call)
		}
	}
	if len(dw.calls) != 1 {
		t.Fatalf("derived parent writes = %d, want 1", len(dw.calls))
	}
	for index := range filters {
		fieldAPI := "Count" + string(rune('A'+index))
		if got := dw.calls[0].updates[fieldAPI]; got != float64(11) {
			t.Fatalf("%s = %#v, want 11", fieldAPI, got)
		}
	}
}

// TestCountUpdateDelta_ReparentMovesOneFromOldToNew verifies the canonical
// relink case: an unfiltered COUNT child reparents from P1 to P2, decrementing
// P1 by 1 and incrementing P2 by 1 — without re-COUNTing either parent.
func TestCountUpdateDelta_ReparentMovesOneFromOldToNew(t *testing.T) {
	const (
		parentObj   = "Campaign"
		childObj    = "CampaignMember"
		rollupField = "NumberOfContacts"
		relField    = "CampaignId"
		oldPID      = "p-001"
		newPID      = "p-002"
		childID     = "cm-1"
	)

	meta := newFakeMetadata(countSchema(parentObj, childObj, rollupField, relField, nil))
	repo := &stubRollupCalcRepo{calcValue: float64(0)} // nil filter → always matches
	svc := NewRollupService(repo, meta, alwaysPessimisticRollupTxManager{}, noopLogger())
	dw := &stubDerivedWriteCapture{rows: map[string]models.SObject{
		parentObj + "|" + oldPID: {rollupField: float64(10)},
		parentObj + "|" + newPID: {rollupField: float64(5)},
	}}
	svc.SetDerivedWriter(dw)

	finalRecord := models.SObject{constants.FieldID: childID, relField: newPID}
	oldRecord := models.SObject{constants.FieldID: childID, relField: oldPID}

	ctx, err := svc.StageRollupCountUpdatePreimage(context.Background(), childObj, oldRecord)
	if err != nil {
		t.Fatalf("StageRollupCountUpdatePreimage error: %v", err)
	}
	if err := svc.ProcessRollupsForUpdate(ctx, childObj, finalRecord, oldRecord); err != nil {
		t.Fatalf("ProcessRollupsForUpdate error: %v", err)
	}

	if repo.currentReadCalls != 0 || repo.calculateCalls != 0 {
		t.Fatalf("delta path should not recompute: currentRead=%d calculate=%d", repo.currentReadCalls, repo.calculateCalls)
	}
	got := map[string]interface{}{}
	for _, c := range dw.calls {
		got[c.recordID] = c.updates[rollupField]
	}
	if len(dw.calls) != 2 {
		t.Fatalf("want 2 parent updates, got %d (%v)", len(dw.calls), got)
	}
	if got[oldPID] != float64(9) {
		t.Errorf("old parent count = %v, want 9 (10-1)", got[oldPID])
	}
	if got[newPID] != float64(6) {
		t.Errorf("new parent count = %v, want 6 (5+1)", got[newPID])
	}
}

// TestCountUpdateDelta_SameParentFilterFlipPlusOne verifies a filtered COUNT
// where the child stays on one parent but its filter-match flips false→true:
// the parent count increments by exactly 1.
func TestCountUpdateDelta_SameParentFilterFlipPlusOne(t *testing.T) {
	const (
		parentObj   = "Campaign"
		childObj    = "CampaignMember"
		rollupField = "NumberOfResponses"
		relField    = "CampaignId"
		pid         = "p-001"
		childID     = "cm-1"
	)
	filter := "HasResponded = 1"

	meta := newFakeMetadata(countSchema(parentObj, childObj, rollupField, relField, &filter))
	// Call 1 (preimage probe) = false (old did not match); call 2 (newMatch) = true.
	repo := &seqMatchRepo{matchSeq: []bool{false, true}}
	svc := NewRollupService(repo, meta, alwaysPessimisticRollupTxManager{}, noopLogger())
	dw := &stubDerivedWriteCapture{rows: map[string]models.SObject{
		parentObj + "|" + pid: {rollupField: float64(7)},
	}}
	svc.SetDerivedWriter(dw)

	finalRecord := models.SObject{constants.FieldID: childID, relField: pid}
	oldRecord := models.SObject{constants.FieldID: childID, relField: pid}

	ctx, err := svc.StageRollupCountUpdatePreimage(context.Background(), childObj, oldRecord)
	if err != nil {
		t.Fatalf("StageRollupCountUpdatePreimage error: %v", err)
	}
	if err := svc.ProcessRollupsForUpdate(ctx, childObj, finalRecord, oldRecord); err != nil {
		t.Fatalf("ProcessRollupsForUpdate error: %v", err)
	}

	if len(dw.calls) != 1 {
		t.Fatalf("want 1 parent update, got %d", len(dw.calls))
	}
	if dw.calls[0].updates[rollupField] != float64(8) {
		t.Errorf("count = %v, want 8 (7+1)", dw.calls[0].updates[rollupField])
	}
}

// TestCountUpdateDelta_NoChangeSkipsWrite verifies that when the child stays on
// the same parent and its match does not flip, the net delta is zero and no
// parent write is issued (and no recompute occurs).
func TestCountUpdateDelta_NoChangeSkipsWrite(t *testing.T) {
	const (
		parentObj   = "Campaign"
		childObj    = "CampaignMember"
		rollupField = "NumberOfContacts"
		relField    = "CampaignId"
		pid         = "p-001"
		childID     = "cm-1"
	)

	meta := newFakeMetadata(countSchema(parentObj, childObj, rollupField, relField, nil))
	repo := &stubRollupCalcRepo{}
	svc := NewRollupService(repo, meta, alwaysPessimisticRollupTxManager{}, noopLogger())
	dw := &stubDerivedWriteCapture{rows: map[string]models.SObject{
		parentObj + "|" + pid: {rollupField: float64(3)},
	}}
	svc.SetDerivedWriter(dw)

	finalRecord := models.SObject{constants.FieldID: childID, relField: pid}
	oldRecord := models.SObject{constants.FieldID: childID, relField: pid}

	ctx, err := svc.StageRollupCountUpdatePreimage(context.Background(), childObj, oldRecord)
	if err != nil {
		t.Fatalf("StageRollupCountUpdatePreimage error: %v", err)
	}
	if err := svc.ProcessRollupsForUpdate(ctx, childObj, finalRecord, oldRecord); err != nil {
		t.Fatalf("ProcessRollupsForUpdate error: %v", err)
	}

	if len(dw.calls) != 0 {
		t.Fatalf("want 0 parent updates (net delta 0), got %d", len(dw.calls))
	}
	if repo.currentReadCalls != 0 || repo.calculateCalls != 0 {
		t.Fatalf("no-op delta must not recompute: currentRead=%d calculate=%d", repo.currentReadCalls, repo.calculateCalls)
	}
}

// TestCountUpdateDelta_NonCountFallsBackToRecompute verifies fail-closed
// behavior: if any affected rollup is non-COUNT (e.g. SUM), the whole update
// falls back to the full recompute path rather than guessing a delta.
func TestCountUpdateDelta_NonCountFallsBackToRecompute(t *testing.T) {
	const (
		parentObj = "Campaign"
		childObj  = "CampaignMember"
		relField  = "CampaignId"
		pid       = "p-001"
		childID   = "cm-1"
	)

	schemas := map[string]*models.ObjectMetadata{
		childObj: {APIName: childObj},
		parentObj: {
			APIName: parentObj,
			Fields: []models.FieldMetadata{{
				APIName: "TotalAmount",
				Type:    constants.FieldTypeRollupSummary,
				RollupConfig: &models.RollupConfig{
					CalcType:          "SUM",
					SummaryObject:     childObj,
					SummaryField:      "Amount",
					RelationshipField: relField,
				},
			}},
		},
	}
	meta := newFakeMetadata(schemas)
	repo := &stubRollupCalcRepo{calcValue: float64(99)}
	svc := NewRollupService(repo, meta, alwaysPessimisticRollupTxManager{}, noopLogger())
	dw := &stubDerivedWriteCapture{}
	svc.SetDerivedWriter(dw)

	finalRecord := models.SObject{constants.FieldID: childID, relField: pid, "Amount": float64(50)}
	oldRecord := models.SObject{constants.FieldID: childID, relField: pid, "Amount": float64(30)}

	ctx, err := svc.StageRollupCountUpdatePreimage(context.Background(), childObj, oldRecord)
	if err != nil {
		t.Fatalf("StageRollupCountUpdatePreimage error: %v", err)
	}
	if err := svc.ProcessRollupsForUpdate(ctx, childObj, finalRecord, oldRecord); err != nil {
		t.Fatalf("ProcessRollupsForUpdate error: %v", err)
	}

	if repo.currentReadCalls == 0 {
		t.Fatalf("non-COUNT rollup must fall back to recompute (CalculateRollupCurrentRead), got 0 calls")
	}
	if dw.calls[0].updates["TotalAmount"] != float64(99) {
		t.Errorf("SUM recompute value = %v, want 99 from CalculateRollup", dw.calls[0].updates["TotalAmount"])
	}
}

// TestCountUpdateDelta_NoPreimageFallsBackToRecompute verifies that when the
// pre-update snapshot was never staged (ctx carries no preimage), the COUNT
// update path declines to apply deltas and the engine recomputes instead.
func TestCountUpdateDelta_NoPreimageFallsBackToRecompute(t *testing.T) {
	const (
		parentObj   = "Campaign"
		childObj    = "CampaignMember"
		rollupField = "NumberOfContacts"
		relField    = "CampaignId"
		oldPID      = "p-001"
		newPID      = "p-002"
		childID     = "cm-1"
	)

	meta := newFakeMetadata(countSchema(parentObj, childObj, rollupField, relField, nil))
	repo := &stubRollupCalcRepo{calcValue: float64(123)}
	svc := NewRollupService(repo, meta, alwaysPessimisticRollupTxManager{}, noopLogger())
	dw := &stubDerivedWriteCapture{}
	svc.SetDerivedWriter(dw)

	finalRecord := models.SObject{constants.FieldID: childID, relField: newPID}
	oldRecord := models.SObject{constants.FieldID: childID, relField: oldPID}

	// No StageRollupCountUpdatePreimage call — ctx has no snapshot.
	if err := svc.ProcessRollupsForUpdate(context.Background(), childObj, finalRecord, oldRecord); err != nil {
		t.Fatalf("ProcessRollupsForUpdate error: %v", err)
	}

	if repo.currentReadCalls == 0 {
		t.Fatalf("missing preimage must fall back to recompute, got 0 CalculateRollupCurrentRead calls")
	}
	if len(dw.calls) != 2 {
		t.Fatalf("reparent fallback updates = %d, want 2", len(dw.calls))
	}
	for _, call := range dw.calls {
		if call.updates[rollupField] != float64(123) {
			t.Errorf("recompute value for parent %s = %v, want 123", call.recordID, call.updates[rollupField])
		}
	}
}