package controlplane

import (
	"strings"
	"testing"

	wq "github.com/pingcap-inc/nexuscrm-connectors/features/connector/workqueue"
)

func rangeProofIdentity() wq.SourceProjectionIdentity {
	return wq.SourceProjectionIdentity{InstalledPackID: "range-proof-pack", EventKind: "record.changed", RouteVersion: 1, WorkType: wq.WorkTypeConnectorEvent, SourceLaneKey: "range-proof-source"}
}
func rangeProofEvent(id string, ordinal int64) *projectionCutoverEvent {
	return &projectionCutoverEvent{id: id, ordinal: ordinal, identity: rangeProofIdentity(), projection: true}
}
func rangeProofWork(id string, e *projectionCutoverEvent, state string) *projectionCutoverWork {
	return &projectionCutoverWork{id: "csp:" + id, eventID: e.id, key: wq.SourceProjectionKey(e.identity), identity: e.identity, ordinal: e.ordinal, state: state, authoritative: true, terminal: state == "processed" || state == "skipped" || state == "dead"}
}

func TestProjectionRangeCutoverProof(t *testing.T) {
	for _, scenario := range []string{"empty", "positive_completed", "closed_mixed_history", "unresolved_mixed_history", "ordinary_terminal_is_not_projection_coverage", "negative_ordinal", "duplicate_ordinal", "ordinary_pending_is_preserved", "event_owned_terminal"} {
		t.Run(scenario, func(t *testing.T) {
			a, b := rangeProofEvent("input", 1), rangeProofEvent("head", 2)
			es := map[string]*projectionCutoverEvent{a.id: a, b.id: b}
			ws := []*projectionCutoverWork{rangeProofWork("head", b, "processed")}
			want := int64(2)
			wantErr := ""
			switch scenario {
			case "empty":
				es = nil
				ws = nil
				want = 0
			case "closed_mixed_history":
				a.ordinal = 0
				a.closed = true
			case "unresolved_mixed_history":
				a.ordinal = 0
				wantErr = "unresolved historical responsibility"
			case "ordinary_terminal_is_not_projection_coverage":
				ws[0].id = "cwi:head"
				ws[0].key = ""
				wantErr = "unresolved historical responsibility"
			case "negative_ordinal":
				a.ordinal = -1
				wantErr = "negative ordinal"
			case "duplicate_ordinal":
				a.ordinal = 2
				wantErr = "duplicate Event ordinal"
			case "ordinary_pending_is_preserved":
				c := rangeProofEvent("command", 50)
				c.identity.EventKind = "command.execute"
				c.projection = false
				es[c.id] = c
				w := rangeProofWork("command", c, "pending")
				w.id = "cwi:command"
				w.key = ""
				ws = append(ws, w)
				want = 50
			case "event_owned_terminal":
				ws = nil
				b.closed = true
			}
			got, err := auditDrainedProjectionHistory(es, ws)
			if wantErr != "" {
				if err == nil || !strings.Contains(err.Error(), wantErr) {
					t.Fatalf("audit=%d/%v want %q", got, err, wantErr)
				}
				return
			}
			if err != nil || got != want {
				t.Fatalf("audit=%d/%v want cutoff %d", got, err, want)
			}
		})
	}
}