package flowcommandv3

import (
	"bytes"
	"strings"
	"testing"

	applicationresult "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/applicationresultv1"
	command "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commandv1"
	flowcommandv1 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv1"
	flowcommandv2 "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/flowcommandv2"
)

func TestPayloadCanonicalRoundTripAndFrozenInputDigest(t *testing.T) {
	payload := validPayload(t, AllowExternalEffects)
	const wantInputDigest = "sha256:02c0d58224905223a336fb98a8fb37f5d36d26da88ffdee3e99d21b1a87a3b71"
	if payload.ExecutionRef.InputDigest != wantInputDigest {
		t.Fatalf("input digest = %q, want %q", payload.ExecutionRef.InputDigest, wantInputDigest)
	}
	raw, err := Encode(payload)
	if err != nil {
		t.Fatalf("Encode: %v", err)
	}
	want := []byte(
		`{"canonicalInputs":{"record":{"id":"001"},"retry":false},"executionRef":{"definitionRef":"flow.neutral","definitionVersion":"flow-version-7","executionId":"execution-1","inputDigest":"` + wantInputDigest + `","packGenerationId":"pack-generation-1","servingEpochDigest":"sha256:` + strings.Repeat("1", 64) + `"},"externalEffectsMode":"ALLOW_EXTERNAL_EFFECTS"}`,
	)
	if !bytes.Equal(raw, want) {
		t.Fatalf("payload = %s, want %s", raw, want)
	}
	decoded, err := Decode(raw)
	if err != nil {
		t.Fatalf("Decode: %v", err)
	}
	if decoded.ExecutionRef != payload.ExecutionRef ||
		decoded.ExternalEffectsMode != payload.ExternalEffectsMode ||
		!bytes.Equal(decoded.CanonicalInputs, payload.CanonicalInputs) {
		t.Fatalf("decoded = %#v, want %#v", decoded, payload)
	}
}

func TestPayloadBindsEveryExecutionAndInputFact(t *testing.T) {
	base := validPayload(t, AllowExternalEffects)
	baseRaw, err := Encode(base)
	if err != nil {
		t.Fatal(err)
	}
	baseDigest := command.ComputePayloadDigest(baseRaw)

	refChanges := map[string]func(*applicationresult.ExecutionRef){
		"execution":  func(ref *applicationresult.ExecutionRef) { ref.ExecutionID = "execution-2" },
		"generation": func(ref *applicationresult.ExecutionRef) { ref.PackGenerationID = "pack-generation-2" },
		"epoch":      func(ref *applicationresult.ExecutionRef) { ref.ServingEpochDigest = digestOf("epoch-2") },
		"definition": func(ref *applicationresult.ExecutionRef) { ref.DefinitionRef = "flow.other" },
		"version":    func(ref *applicationresult.ExecutionRef) { ref.DefinitionVersion = "flow-version-8" },
	}
	for name, change := range refChanges {
		t.Run(name, func(t *testing.T) {
			changed := clonePayload(base)
			change(&changed.ExecutionRef)
			raw, err := Encode(changed)
			if err != nil || command.ComputePayloadDigest(raw) == baseDigest {
				t.Fatalf("changed payload digest did not change: %v", err)
			}
		})
	}

	for name, mutate := range map[string]func(*Payload){
		"inputs": func(payload *Payload) {
			payload.CanonicalInputs = []byte(`{"record":{"id":"002"},"retry":false}`)
		},
		"mode": func(payload *Payload) {
			payload.ExternalEffectsMode = SuppressExternalEffects
		},
	} {
		t.Run(name, func(t *testing.T) {
			changed := clonePayload(base)
			mutate(&changed)
			changed.ExecutionRef.InputDigest = mustInputDigest(
				t, changed.CanonicalInputs, changed.ExternalEffectsMode,
			)
			raw, err := Encode(changed)
			if err != nil || command.ComputePayloadDigest(raw) == baseDigest {
				t.Fatalf("changed payload digest did not change: %v", err)
			}
		})
	}
}

func TestExternalEffectsModesAreExplicitAndDisjoint(t *testing.T) {
	allowed := validPayload(t, AllowExternalEffects)
	suppressed := validPayload(t, SuppressExternalEffects)
	if allowed.ExecutionRef.InputDigest == suppressed.ExecutionRef.InputDigest {
		t.Fatal("external-effects modes produced the same input digest")
	}
	for _, payload := range []Payload{allowed, suppressed} {
		raw, err := Encode(payload)
		if err != nil {
			t.Fatalf("Encode(%q): %v", payload.ExternalEffectsMode, err)
		}
		decoded, err := Decode(raw)
		if err != nil || decoded.ExternalEffectsMode != payload.ExternalEffectsMode {
			t.Fatalf("Decode(%q) = %#v, %v", payload.ExternalEffectsMode, decoded, err)
		}
	}
	for _, mode := range []ExternalEffectsMode{"", "DEFAULT", "suppress_external_effects"} {
		payload := validPayload(t, AllowExternalEffects)
		payload.ExternalEffectsMode = mode
		if _, err := Encode(payload); err == nil {
			t.Fatalf("Encode accepted mode %q", mode)
		}
		if digest, err := ComputeInputDigest(payload.CanonicalInputs, mode); err == nil || digest != "" {
			t.Fatalf("ComputeInputDigest(%q) = %q, %v", mode, digest, err)
		}
	}
}

func TestPayloadRejectsStaleOrForgedInputDigest(t *testing.T) {
	payload := validPayload(t, AllowExternalEffects)
	payload.ExecutionRef.InputDigest = digestOf("forged")
	if _, err := Encode(payload); err == nil {
		t.Fatal("Encode accepted a forged input digest")
	}

	valid := validPayload(t, AllowExternalEffects)
	raw, err := Encode(valid)
	if err != nil {
		t.Fatal(err)
	}
	changedInput := bytes.Replace(
		raw, []byte(`"id":"001"`), []byte(`"id":"002"`), 1,
	)
	if _, err := Decode(changedInput); err == nil {
		t.Fatal("Decode accepted changed inputs with a stale digest")
	}
	changedMode := bytes.Replace(
		raw, []byte(AllowExternalEffects), []byte(SuppressExternalEffects), 1,
	)
	if _, err := Decode(changedMode); err == nil {
		t.Fatal("Decode accepted a changed mode with a stale digest")
	}
}

func TestPayloadVersionBoundariesAreFailClosed(t *testing.T) {
	if CommandKind != flowcommandv1.CommandKind ||
		PayloadFormatVersion == flowcommandv1.PayloadFormatVersion ||
		PayloadFormatVersion == flowcommandv2.PayloadFormatVersion {
		t.Fatalf("command/version = %q/%d", CommandKind, PayloadFormatVersion)
	}
	v1, err := flowcommandv1.Encode(flowcommandv1.Payload{
		DefinitionRef: "flow.neutral", DefinitionVersion: "flow-version-7",
		CanonicalInputs: []byte(`{}`),
	})
	if err != nil {
		t.Fatal(err)
	}
	v2, err := flowcommandv2.Encode(flowcommandv2.Payload{
		DefinitionRef: "flow.neutral", DefinitionVersion: "flow-version-7",
		CanonicalInputs: []byte(`{}`),
		DependencySource: flowcommandv2.DependencySourceScope{
			ObjectKind: "records", SourceID: "source-1",
		},
	})
	if err != nil {
		t.Fatal(err)
	}
	v3, err := Encode(validPayload(t, AllowExternalEffects))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := Decode(v1); err == nil {
		t.Fatal("v3 decoder accepted v1")
	}
	if _, err := Decode(v2); err == nil {
		t.Fatal("v3 decoder accepted v2")
	}
	if _, err := flowcommandv1.Decode(v3); err == nil {
		t.Fatal("v1 decoder accepted v3")
	}
	if _, err := flowcommandv2.Decode(v3); err == nil {
		t.Fatal("v2 decoder accepted v3")
	}
}

func TestPayloadDecoderFailsClosed(t *testing.T) {
	valid, err := Encode(validPayload(t, AllowExternalEffects))
	if err != nil {
		t.Fatal(err)
	}
	validString := string(valid)
	refEnd := `"servingEpochDigest":"sha256:` + strings.Repeat("1", 64) + `"}`
	for name, raw := range map[string][]byte{
		"noncanonical input":  bytes.Replace(valid, []byte(`{"record":{"id":"001"},"retry":false}`), []byte(`{"retry":false,"record":{"id":"001"}}`), 1),
		"unknown top field":   []byte(strings.TrimSuffix(validString, `}`) + `,"z":true}`),
		"duplicate top field": []byte(strings.Replace(validString, `"externalEffectsMode":`, `"externalEffectsMode":"ALLOW_EXTERNAL_EFFECTS","externalEffectsMode":`, 1)),
		"missing top field":   []byte(strings.Replace(validString, `,"externalEffectsMode":"ALLOW_EXTERNAL_EFFECTS"`, ``, 1)),
		"case variant":        []byte(strings.Replace(validString, `"executionRef"`, `"ExecutionRef"`, 1)),
		"unknown ref field":   []byte(strings.Replace(validString, refEnd, strings.TrimSuffix(refEnd, `}`)+`,"z":true}`, 1)),
		"duplicate ref field": []byte(strings.Replace(validString, `"executionId":`, `"executionId":"execution-1","executionId":`, 1)),
		"missing ref field":   []byte(strings.Replace(validString, `"executionId":"execution-1",`, ``, 1)),
		"invalid ref":         []byte(strings.Replace(validString, `"executionId":"execution-1"`, `"executionId":" execution-1"`, 1)),
		"unknown mode":        []byte(strings.Replace(validString, string(AllowExternalEffects), `UNKNOWN`, 1)),
		"scalar inputs":       []byte(strings.Replace(validString, `{"record":{"id":"001"},"retry":false}`, `1`, 1)),
		"trailing":            append(append([]byte(nil), valid...), []byte(` {}`)...),
		"invalid utf8":        []byte{0xff},
	} {
		t.Run(name, func(t *testing.T) {
			if _, err := Decode(raw); err == nil {
				t.Fatalf("Decode succeeded: %s", raw)
			}
		})
	}
	if _, err := Decode(make([]byte, MaxPayloadBytes+1)); err == nil {
		t.Fatal("Decode accepted an oversized payload")
	}
	oversizedInputs := []byte(`{"value":"` + strings.Repeat("x", MaxPayloadBytes) + `"}`)
	if digest, err := ComputeInputDigest(oversizedInputs, AllowExternalEffects); err == nil || digest != "" {
		t.Fatalf("oversized input digest = %q, %v", digest, err)
	}
}

func validPayload(t *testing.T, mode ExternalEffectsMode) Payload {
	t.Helper()
	inputs := []byte(`{"record":{"id":"001"},"retry":false}`)
	return Payload{
		ExecutionRef: applicationresult.ExecutionRef{
			ExecutionID: "execution-1", PackGenerationID: "pack-generation-1",
			ServingEpochDigest: "sha256:" + strings.Repeat("1", 64), DefinitionRef: "flow.neutral",
			DefinitionVersion: "flow-version-7", InputDigest: mustInputDigest(t, inputs, mode),
		},
		CanonicalInputs: inputs, ExternalEffectsMode: mode,
	}
}

func clonePayload(payload Payload) Payload {
	payload.CanonicalInputs = append([]byte(nil), payload.CanonicalInputs...)
	return payload
}

func mustInputDigest(t *testing.T, inputs []byte, mode ExternalEffectsMode) string {
	t.Helper()
	digest, err := ComputeInputDigest(inputs, mode)
	if err != nil {
		t.Fatal(err)
	}
	return digest
}

func digestOf(value string) string {
	return command.ComputePayloadDigest([]byte(value))
}