package filesv1

import (
	"context"
	"strings"
	"testing"
	"time"

	common "github.com/pingcap-inc/nexuscrm-contract/pkg/runtime/commonv1"
)

func TestCreateStreamStateOwnsAggregateContract(t *testing.T) {
	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(time.Second))
	defer cancel()
	state, err := NewCreateStreamState(ctx)
	if err != nil {
		t.Fatal(err)
	}
	header := CreateHeader{IdempotencyKey: "request-1", ParentObjectID: "Account",
		ParentRecordID: "001", FileName: "file.txt", ContentType: "text/plain", ContentLength: 3}
	if err := state.Observe(CreateChunk{Header: &header}); err != nil {
		t.Fatalf("Observe(header) error = %v", err)
	}
	if err := state.Observe(CreateChunk{Data: []byte("abc")}); err != nil {
		t.Fatalf("Observe(data) error = %v", err)
	}
	if err := state.Complete(); err != nil {
		t.Fatalf("Complete() error = %v", err)
	}
	overflow, err := NewCreateStreamState(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if err := overflow.Observe(CreateChunk{Header: &header}); err != nil {
		t.Fatal(err)
	}
	if overflow.Observe(CreateChunk{Data: []byte("abcd")}) == nil {
		t.Fatal("content longer than declared was accepted")
	}
	dataFirst, err := NewCreateStreamState(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if dataFirst.Observe(CreateChunk{Data: []byte("a")}) == nil {
		t.Fatal("data before the create header was accepted")
	}
	duplicateHeader, err := NewCreateStreamState(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if duplicateHeader.Observe(CreateChunk{Header: &header}) != nil ||
		duplicateHeader.Observe(CreateChunk{Header: &header}) == nil {
		t.Fatal("duplicate create header was accepted")
	}
	incomplete, err := NewCreateStreamState(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if incomplete.Observe(CreateChunk{Header: &header}) != nil ||
		incomplete.Observe(CreateChunk{Data: []byte("a")}) != nil ||
		incomplete.Complete() == nil {
		t.Fatal("incomplete create stream was committed")
	}
	canceledContext, cancelUpload := context.WithDeadline(context.Background(), time.Now().Add(time.Second))
	canceled, err := NewCreateStreamState(canceledContext)
	if err != nil {
		t.Fatal(err)
	}
	if err := canceled.Observe(CreateChunk{Header: &header}); err != nil {
		t.Fatal(err)
	}
	cancelUpload()
	if canceled.Observe(CreateChunk{Data: []byte("abc")}) == nil || canceled.Complete() == nil {
		t.Fatal("canceled create stream remained committable")
	}
}

func TestFileHeadersAndOpenStreamsAreBounded(t *testing.T) {
	header := CreateHeader{IdempotencyKey: "request-1", ParentObjectID: "Account", ParentRecordID: "001",
		FileName: "file.txt", ContentType: "text/plain", AttachmentKind: strings.Repeat("x", MaxIdentifierBytes+1)}
	if ValidateCreateHeader(header) == nil {
		t.Fatal("oversized attachment kind was accepted")
	}
	var state OpenStreamState
	receipts := common.Receipts{ProtocolVersion: "1.0", CapabilityVersion: CapabilityName}
	if err := state.Observe(OpenChunk{FileName: "file.txt", ContentType: "text/plain", TotalSize: 1, Data: []byte("x"), Receipts: &receipts}); err != nil {
		t.Fatalf("Observe(open) error = %v", err)
	}
	if err := state.Complete(); err != nil {
		t.Fatalf("Complete(open) error = %v", err)
	}
	var drift OpenStreamState
	if err := drift.Observe(OpenChunk{FileName: "file.txt", ContentType: "text/plain", TotalSize: 2, Data: []byte("x"), Receipts: &receipts}); err != nil {
		t.Fatal(err)
	}
	if drift.Observe(OpenChunk{FileName: "other.txt", ContentType: "text/plain", TotalSize: 2, Data: []byte("y")}) == nil {
		t.Fatal("open stream metadata drift was accepted")
	}
	var lateReceipts OpenStreamState
	if err := lateReceipts.Observe(OpenChunk{FileName: "file.txt", ContentType: "text/plain", TotalSize: 2, Data: []byte("x"), Receipts: &receipts}); err != nil {
		t.Fatal(err)
	}
	if lateReceipts.Observe(OpenChunk{FileName: "file.txt", ContentType: "text/plain", TotalSize: 2, Data: []byte("y"), Receipts: &receipts}) == nil {
		t.Fatal("receipts after the first open frame were accepted")
	}
}