package automation

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"reflect"
	"strings"
	"testing"

	"github.com/nexuscrm/kernel/pkg/logging"
	"github.com/nexuscrm/shared/pkg/api"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
	"github.com/rs/zerolog"
)

type scheduledInputQueryCall struct {
	objectName string
	filterExpr string
	user       *models.UserSession
	sortField  string
	sortDir    string
	limit      int
}

type scheduledInputQueryStub struct {
	calls     []scheduledInputQueryCall
	results   []*models.QueryResult
	errs      []error
	responder func(callIndex int, call scheduledInputQueryCall) (*models.QueryResult, error)
}

func (q *scheduledInputQueryStub) QueryWithFilter(ctx context.Context, objectName string, filterExpr string, user *models.UserSession, sortField string, sortDir string, limit int) (*models.QueryResult, error) {
	call := scheduledInputQueryCall{
		objectName: objectName,
		filterExpr: filterExpr,
		user:       user,
		sortField:  sortField,
		sortDir:    sortDir,
		limit:      limit,
	}
	q.calls = append(q.calls, call)
	callIndex := len(q.calls) - 1

	if q.responder != nil {
		return q.responder(callIndex, call)
	}
	if callIndex < len(q.errs) && q.errs[callIndex] != nil {
		return nil, q.errs[callIndex]
	}
	if callIndex < len(q.results) && q.results[callIndex] != nil {
		return q.results[callIndex], nil
	}
	return &models.QueryResult{}, nil
}

func (q *scheduledInputQueryStub) Query(context.Context, api.QueryRequest, *models.UserSession) (*models.QueryResult, error) {
	return nil, nil
}

func (q *scheduledInputQueryStub) QueryWithSpec(context.Context, queryspec.QuerySpec, *models.UserSession) (*models.QueryResult, error) {
	return nil, nil
}

func (q *scheduledInputQueryStub) RunAnalytics(context.Context, models.AnalyticsQuery, *models.UserSession) (interface{}, error) {
	return nil, nil
}

func (q *scheduledInputQueryStub) SearchSingleObject(context.Context, string, string, []string, []string, int, int, string, string, string, *models.UserSession) ([]models.SObject, error) {
	return nil, nil
}

func (q *scheduledInputQueryStub) SearchSuggestions(context.Context, string, string, []string, []string, int, *models.UserSession) ([]models.SObject, error) {
	return nil, nil
}

func (q *scheduledInputQueryStub) GlobalSearch(context.Context, string, *models.UserSession) ([]models.SearchResult, error) {
	return nil, nil
}

func (q *scheduledInputQueryStub) Calculate(context.Context, string, models.SObject, *models.UserSession) (models.SObject, error) {
	return nil, nil
}

func (q *scheduledInputQueryStub) ExecuteRawSQL(context.Context, string, []interface{}, *models.UserSession) ([]models.SObject, error) {
	return nil, nil
}

func buildScheduledFlowVersionForTest(filters []models.FlowRecordFilter, logic string) *models.FlowVersion {
	return &models.FlowVersion{
		Metadata: models.FlowMetadata{
			Start: &models.FlowStart{
				Object:      "GenericObject__c",
				FilterLogic: logic,
				Filters:     filters,
			},
		},
	}
}

func scheduledInputIDs(inputs []map[string]interface{}) []string {
	ids := make([]string, 0, len(inputs))
	for _, input := range inputs {
		record, ok := input["$Record"].(models.SObject)
		if !ok {
			continue
		}
		ids = append(ids, record.GetString("Id"))
	}
	return ids
}

func makeStatusPage(start, end int, status string) []models.SObject {
	records := make([]models.SObject, 0, end-start+1)
	for i := start; i <= end; i++ {
		records = append(records, models.SObject{
			"Id":     fmt.Sprintf("%04d", i),
			"Status": status,
		})
	}
	return records
}

func newScheduledInputTestLogger(buf *bytes.Buffer) *logging.Logger {
	tctx := zerolog.New(buf).WithContext(context.Background())
	return logging.New(tctx)
}

func TestBuildScheduledRunInputs_AndFallbackDropsWholeAggregate(t *testing.T) {
	query := &scheduledInputQueryStub{
		results: []*models.QueryResult{{
			Records: []models.SObject{{"Id": "0001", "Status": "Open"}},
		}},
	}
	executor := &FlowExecutor{query: query}

	got, err := executor.buildScheduledRunInputs(context.Background(), buildScheduledFlowVersionForTest([]models.FlowRecordFilter{
		{Field: "Status", Operator: "EqualTo", Value: stringValue("Open")},
		{Field: "Status", Operator: "In", Value: []interface{}{"Open", "Pending"}},
	}, "AND"), nil)
	if err != nil {
		t.Fatalf("buildScheduledRunInputs() error = %v", err)
	}
	if len(query.calls) != 1 {
		t.Fatalf("QueryWithFilter() calls = %d, want 1", len(query.calls))
	}
	if query.calls[0].filterExpr != "" {
		t.Fatalf("QueryWithFilter() filter = %q, want empty aggregate when any filter falls back", query.calls[0].filterExpr)
	}
	if query.calls[0].sortField != "Id" || query.calls[0].sortDir != "ASC" || query.calls[0].limit != 500 {
		t.Fatalf("QueryWithFilter() paging = (%q, %q, %d), want (\"Id\", \"ASC\", 500)", query.calls[0].sortField, query.calls[0].sortDir, query.calls[0].limit)
	}
	if ids := scheduledInputIDs(got); !reflect.DeepEqual(ids, []string{"0001"}) {
		t.Fatalf("scheduledInputIDs = %v, want [0001]", ids)
	}
}

func TestBuildScheduledRunInputs_SQLPrefilterMatchesInMemoryGate(t *testing.T) {
	query := &scheduledInputQueryStub{
		results: []*models.QueryResult{{
			Records: []models.SObject{
				{"Id": "0001", "Status": "Open"},
				{"Id": "0002", "Status": "Closed"},
				{"Id": "0003", "Status": "Open"},
			},
		}},
	}
	executor := &FlowExecutor{query: query}
	user := &models.UserSession{ID: "automation-user"}

	got, err := executor.buildScheduledRunInputs(context.Background(), buildScheduledFlowVersionForTest([]models.FlowRecordFilter{
		{Field: "Status", Operator: "EqualTo", Value: stringValue("Open")},
	}, "AND"), user)
	if err != nil {
		t.Fatalf("buildScheduledRunInputs() error = %v", err)
	}
	if len(query.calls) != 1 {
		t.Fatalf("QueryWithFilter() calls = %d, want 1", len(query.calls))
	}
	call := query.calls[0]
	if call.objectName != "GenericObject__c" {
		t.Fatalf("QueryWithFilter() object = %q, want %q", call.objectName, "GenericObject__c")
	}
	if call.filterExpr != "Status = 'Open'" {
		t.Fatalf("QueryWithFilter() filter = %q, want %q", call.filterExpr, "Status = 'Open'")
	}
	if call.user != user {
		t.Fatalf("QueryWithFilter() user = %#v, want %#v", call.user, user)
	}
	if call.sortField != "Id" || call.sortDir != "ASC" || call.limit != 500 {
		t.Fatalf("QueryWithFilter() paging = (%q, %q, %d), want (\"Id\", \"ASC\", 500)", call.sortField, call.sortDir, call.limit)
	}
	if ids := scheduledInputIDs(got); !reflect.DeepEqual(ids, []string{"0001", "0003"}) {
		t.Fatalf("scheduledInputIDs = %v, want [0001 0003]", ids)
	}
	record, ok := got[0]["$Record"].(models.SObject)
	if !ok {
		t.Fatalf("$Record type = %T, want models.SObject", got[0]["$Record"])
	}
	if got := models.RecordObjectType(record); got != "GenericObject__c" {
		t.Fatalf("$Record object type = %q, want %q", got, "GenericObject__c")
	}
	if _, ok := record["attributes"]; ok {
		t.Fatalf("$Record emitted v63 attributes in automation layer: %#v", record["attributes"])
	}
}

func TestBuildScheduledRunInputs_IncludesExcludesFallback_PreservesRuntimeStorageSemantics(t *testing.T) {
	tests := []struct {
		name     string
		operator string
		wantIDs  []string
	}{
		{name: "Includes", operator: "Includes", wantIDs: []string{"0001", "0002"}},
		{name: "Excludes", operator: "Excludes", wantIDs: []string{"0003", "0004"}},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			query := &scheduledInputQueryStub{
				results: []*models.QueryResult{{
					Records: []models.SObject{
						{"Id": "0001", "Subject": `["Meeting","Email"]`},
						{"Id": "0002", "Subject": "Phone;Meeting"},
						{"Id": "0003", "Subject": `["Phone","Fax"]`},
						{"Id": "0004", "Subject": "Phone;Fax"},
					},
				}},
			}
			executor := &FlowExecutor{query: query}

			got, err := executor.buildScheduledRunInputs(context.Background(), buildScheduledFlowVersionForTest([]models.FlowRecordFilter{
				{Field: "Subject", Operator: tt.operator, Value: stringValue("Meeting;Email")},
			}, "AND"), nil)
			if err != nil {
				t.Fatalf("buildScheduledRunInputs() error = %v", err)
			}
			if len(query.calls) != 1 {
				t.Fatalf("QueryWithFilter() calls = %d, want 1", len(query.calls))
			}
			if query.calls[0].filterExpr != "" {
				t.Fatalf("QueryWithFilter() filter = %q, want empty aggregate because scheduled-start %s must fail closed", query.calls[0].filterExpr, tt.operator)
			}
			if ids := scheduledInputIDs(got); !reflect.DeepEqual(ids, tt.wantIDs) {
				t.Fatalf("scheduledInputIDs = %v, want %v", ids, tt.wantIDs)
			}
		})
	}
}

func TestBuildScheduledRunInputs_BroadFallback_RespectsMaxMatched(t *testing.T) {
	query := &scheduledInputQueryStub{
		responder: func(callIndex int, call scheduledInputQueryCall) (*models.QueryResult, error) {
			switch callIndex {
			case 0:
				return &models.QueryResult{Records: makeStatusPage(1, 500, "Open")}, nil
			case 1:
				if call.filterExpr != "Id > '0500'" {
					t.Fatalf("page 2 filter = %q, want %q", call.filterExpr, "Id > '0500'")
				}
				return &models.QueryResult{Records: makeStatusPage(501, 1000, "Open")}, nil
			default:
				t.Fatalf("unexpected extra query call %d: %#v", callIndex+1, call)
				return nil, nil
			}
		},
	}
	executor := &FlowExecutor{query: query}

	got, err := executor.buildScheduledRunInputs(context.Background(), buildScheduledFlowVersionForTest([]models.FlowRecordFilter{
		{Field: "Status", Operator: "In", Value: []interface{}{"Open"}},
	}, "AND"), nil)
	if err != nil {
		t.Fatalf("buildScheduledRunInputs() error = %v", err)
	}
	if len(query.calls) != 2 {
		t.Fatalf("QueryWithFilter() calls = %d, want 2 paged calls to reach 1000 matches", len(query.calls))
	}
	for i, call := range query.calls {
		if call.sortField != "Id" || call.sortDir != "ASC" || call.limit != 500 {
			t.Fatalf("call %d paging = (%q, %q, %d), want (\"Id\", \"ASC\", 500)", i+1, call.sortField, call.sortDir, call.limit)
		}
	}
	if len(got) != 1000 {
		t.Fatalf("len(inputs) = %d, want 1000 capped matches", len(got))
	}
	if ids := scheduledInputIDs(got[:3]); !reflect.DeepEqual(ids, []string{"0001", "0002", "0003"}) {
		t.Fatalf("first ids = %v, want [0001 0002 0003]", ids)
	}
}

func TestBuildScheduledRunInputs_BroadFallback_LogsStructuredMatchedCapWarning(t *testing.T) {
	query := &scheduledInputQueryStub{
		responder: func(callIndex int, call scheduledInputQueryCall) (*models.QueryResult, error) {
			switch callIndex {
			case 0:
				return &models.QueryResult{Records: makeStatusPage(1, 500, "Open")}, nil
			case 1:
				return &models.QueryResult{Records: makeStatusPage(501, 1000, "Open")}, nil
			default:
				t.Fatalf("unexpected extra query call %d: %#v", callIndex+1, call)
				return nil, nil
			}
		},
	}
	var logBuf bytes.Buffer
	executor := &FlowExecutor{
		query:  query,
		logger: newScheduledInputTestLogger(&logBuf),
	}

	got, err := executor.buildScheduledRunInputs(context.Background(), buildScheduledFlowVersionForTest([]models.FlowRecordFilter{
		{Field: "Status", Operator: "In", Value: []interface{}{"Open"}},
	}, "AND"), nil)
	if err != nil {
		t.Fatalf("buildScheduledRunInputs() error = %v", err)
	}
	if len(got) != scheduledFlowMaxMatched {
		t.Fatalf("len(inputs) = %d, want %d", len(got), scheduledFlowMaxMatched)
	}

	logs := logBuf.String()
	for _, want := range []string{
		`"event":"scheduled_flow_candidate_scan"`,
		`"outcome":"matched_cap_reached"`,
		`"object_name":"GenericObject__c"`,
		`"matched_count":1000`,
		`"scanned_count":1000`,
		`"matched_cap":1000`,
		`"scan_cap":50000`,
		`"sql_prefilter_applied":false`,
	} {
		if !strings.Contains(logs, want) {
			t.Fatalf("matched-cap logs %q missing %q", logs, want)
		}
	}
}

func TestBuildScheduledRunInputs_ScanBudgetExhaustion_FailsClosed(t *testing.T) {
	query := &scheduledInputQueryStub{
		responder: func(callIndex int, call scheduledInputQueryCall) (*models.QueryResult, error) {
			start := callIndex*500 + 1
			end := start + 499
			return &models.QueryResult{Records: makeStatusPage(start, end, "Closed")}, nil
		},
	}
	var logBuf bytes.Buffer
	executor := &FlowExecutor{
		query:  query,
		logger: newScheduledInputTestLogger(&logBuf),
	}

	got, err := executor.buildScheduledRunInputs(context.Background(), buildScheduledFlowVersionForTest([]models.FlowRecordFilter{
		{Field: "Status", Operator: "In", Value: []interface{}{"Open"}},
	}, "AND"), nil)
	if err == nil {
		t.Fatalf("buildScheduledRunInputs() error = nil, want fail-closed scan-budget error; inputs = %#v", got)
	}
	if got != nil {
		t.Fatalf("buildScheduledRunInputs() inputs = %#v, want nil on scan-budget failure", got)
	}
	var scanErr *scheduledFlowCandidateScanBudgetError
	if !errors.As(err, &scanErr) {
		t.Fatalf("buildScheduledRunInputs() error = %T %v, want scheduledFlowCandidateScanBudgetError", err, err)
	}
	if scanErr.ObjectName != "GenericObject__c" {
		t.Fatalf("scanErr.ObjectName = %q, want %q", scanErr.ObjectName, "GenericObject__c")
	}
	if scanErr.ScannedCount != scheduledFlowMaxScanned {
		t.Fatalf("scanErr.ScannedCount = %d, want %d", scanErr.ScannedCount, scheduledFlowMaxScanned)
	}
	if scanErr.MatchedCount != 0 {
		t.Fatalf("scanErr.MatchedCount = %d, want 0", scanErr.MatchedCount)
	}
	if !strings.Contains(logBuf.String(), `"outcome":"scan_budget_exceeded"`) {
		t.Fatalf("scan-budget logs %q missing scan_budget_exceeded outcome", logBuf.String())
	}
	if len(query.calls) != 100 {
		t.Fatalf("QueryWithFilter() calls = %d, want 100 pages to exhaust 50000-row scan budget", len(query.calls))
	}
}

func TestBuildScheduledRunInputs_InvalidMetadataFailsClosed(t *testing.T) {
	query := &scheduledInputQueryStub{
		results: []*models.QueryResult{{
			Records: []models.SObject{{"Id": "0001", "Status": "Open"}},
		}},
	}
	executor := &FlowExecutor{query: query}

	got, err := executor.buildScheduledRunInputs(context.Background(), buildScheduledFlowVersionForTest([]models.FlowRecordFilter{
		{Field: "Bad Field", Operator: "EqualTo", Value: stringValue("Open")},
	}, "AND"), nil)
	if err == nil {
		t.Fatalf("buildScheduledRunInputs() error = nil, want invalid metadata failure; inputs = %#v", got)
	}
	if len(query.calls) != 0 {
		t.Fatalf("QueryWithFilter() calls = %d, want 0 on invalid scheduled-start metadata", len(query.calls))
	}
}