package persistence

import (
	"context"
	"database/sql"
	"errors"
	"testing"
	"time"

	"github.com/nexuscrm/kernel/pkg/schema"
	"github.com/nexuscrm/kernel/pkg/schemacatalog"
)

func TestFileUploadOperationSchemaMatchesCanonicalCatalog(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	definitions, err := schemacatalog.SystemTableDefinitions()
	if err != nil {
		t.Fatal(err)
	}
	wanted := map[string]bool{
		"_System_FileUploadOperation": false,
		"_System_File":                false,
	}
	repository := NewSchemaRepository(db, nil)
	for _, definition := range definitions {
		if _, ok := wanted[definition.TableName]; !ok {
			continue
		}
		candidate := definition
		if definition.TableName == fileUploadOperationTable {
			candidate = legacyFileUploadOperationDefinition(t, definition)
		}
		if err := repository.CreatePhysicalTable(context.Background(), candidate); err != nil {
			t.Fatalf("CreatePhysicalTable(%s), legacy pass: %v", definition.TableName, err)
		}
		if definition.TableName == fileUploadOperationTable {
			requireFileUploadLifecycleIndexMissing(t, db)
			for pass := 1; pass <= 2; pass++ {
				if err := repository.CreatePhysicalTable(context.Background(), definition); err != nil {
					t.Fatalf("CreatePhysicalTable(%s), canonical warm pass %d: %v", definition.TableName, pass, err)
				}
			}
		}
		wanted[definition.TableName] = true
	}
	for table, found := range wanted {
		if !found {
			t.Fatalf("%s absent from canonical schemacatalog", table)
		}
	}
	if err := VerifyFileUploadOperationSchema(
		t.Context(), db,
	); err != nil {
		t.Fatalf("canonical file upload schema rejected: %v", err)
	}
	if err := VerifyExactTableIndex(
		t.Context(), db, fileUploadOperationTable, fileUploadLifecycleIndex,
		false, fileUploadLifecycleIndexColumns,
	); err != nil {
		t.Fatalf("canonical file upload lifecycle index rejected: %v", err)
	}
}

func legacyFileUploadOperationDefinition(t *testing.T, definition schema.TableDefinition) schema.TableDefinition {
	t.Helper()
	legacy := definition
	legacy.Indices = make([]schema.IndexDefinition, 0, len(definition.Indices)-1)
	for _, index := range definition.Indices {
		if index.Name == fileUploadLifecycleIndex {
			continue
		}
		legacy.Indices = append(legacy.Indices, index)
	}
	if len(legacy.Indices) != len(definition.Indices)-1 {
		t.Fatalf("%s is absent from %s", fileUploadLifecycleIndex, definition.TableName)
	}
	return legacy
}

func requireFileUploadLifecycleIndexMissing(t *testing.T, db interface {
	QueryRow(query string, args ...any) *sql.Row
}) {
	t.Helper()
	var count int
	if err := db.QueryRow(
		`SELECT COUNT(*) FROM INFORMATION_SCHEMA.STATISTICS
WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = ? AND INDEX_NAME = ?`,
		fileUploadOperationTable,
		fileUploadLifecycleIndex,
	).Scan(&count); err != nil {
		t.Fatalf("inspect absent file upload lifecycle index: %v", err)
	}
	if count != 0 {
		t.Fatalf("%s.%s unexpectedly exists before canonical warm upgrade", fileUploadOperationTable, fileUploadLifecycleIndex)
	}
}

func TestFileUploadOperationTiDBPersistsRestartAndDigestFences(t *testing.T) {
	db, cleanup := openSessionRepositoryTiDB(t)
	defer cleanup()
	if _, err := db.Exec(`CREATE TABLE _System_FileUploadOperation (
		Id VARCHAR(255) NOT NULL PRIMARY KEY,
		IdempotencyKey VARCHAR(255) NOT NULL,
		RequestHash VARCHAR(255) NOT NULL,
		State VARCHAR(64) NOT NULL DEFAULT 'PREPARED',
		OwnerToken VARCHAR(64) NOT NULL,
		ObjectKey VARCHAR(1024) NULL,
		StorageProvider VARCHAR(64) NULL,
		StorageBucket VARCHAR(255) NULL,
		StorageAuthorityDigest VARCHAR(255) NULL,
		SystemFileId VARCHAR(255) NULL,
		ParentId VARCHAR(255) NOT NULL,
		ParentObjectApiName VARCHAR(255) NOT NULL,
		ErrorCode VARCHAR(255) NULL,
		ErrorMessage TEXT NULL,
		RetryCount INT NOT NULL DEFAULT 0,
		RepairId VARCHAR(255) NULL,
		RepairReason VARCHAR(1024) NULL,
		RepairCount INT NOT NULL DEFAULT 0,
		LastRepairAt DATETIME(6) NULL,
		ExpiresAt DATETIME(6) NULL,
		CreatedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
		LastModifiedDate DATETIME(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6),
		UNIQUE KEY uk__System_FileUploadOperation_idempotency (IdempotencyKey),
		KEY idx__System_FileUploadOperation_expires (ExpiresAt)
	)`); err != nil {
		t.Fatalf("create upload operation table: %v", err)
	}
	if _, err := db.Exec(`CREATE TABLE _System_File (
		Id VARCHAR(255) NOT NULL PRIMARY KEY,
		ParentId VARCHAR(255) NOT NULL,
		ParentObjectApiName VARCHAR(255) NOT NULL,
		Name VARCHAR(255) NOT NULL,
		OriginalFileName VARCHAR(255) NOT NULL,
		MimeType VARCHAR(255) NULL,
		SizeBytes BIGINT NULL,
		StoragePath TEXT NULL,
		StorageKey VARCHAR(1024) NOT NULL,
		StorageProvider VARCHAR(32) NULL,
		StorageBucket VARCHAR(255) NULL,
		Status VARCHAR(32) NOT NULL DEFAULT 'Active',
		SourceType VARCHAR(32) NOT NULL DEFAULT 'native',
		AttachmentKind VARCHAR(64) NOT NULL,
		Checksum VARCHAR(255) NULL
	)`); err != nil {
		t.Fatalf("create canonical file metadata table: %v", err)
	}
	now := time.Now().UTC().Truncate(time.Microsecond)
	input := FileUploadOperationAcceptance{
		ID: "file-1", IdempotencyKey: "file-1",
		RequestHash: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		OwnerToken:  "owner-1", ObjectKey: "Contract/record-1/file-1/proof.txt", SystemFileID: "file-1",
		StorageProvider: "filesystem", StorageBucket: "default",
		StorageAuthorityDigest: "sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
		ParentID:               "record-1", ParentObjectAPIName: "Contract",
		Now: now, LeaseUntil: now.Add(time.Minute),
	}
	repository := NewFileUploadOperationRepository(db)
	if err := VerifyFileUploadOperationSchema(t.Context(), db); err != nil {
		t.Fatalf("verify upload operation schema: %v", err)
	}
	row, owner, err := repository.Accept(t.Context(), input)
	if err != nil || !owner || row.State != FileUploadOperationPrepared {
		t.Fatalf("first Accept() = %#v owner=%t err=%v", row, owner, err)
	}
	if _, err := db.Exec(
		`UPDATE _System_FileUploadOperation SET ExpiresAt = ? WHERE Id = ?`,
		now.Add(-time.Minute), row.ID,
	); err != nil {
		t.Fatal(err)
	}
	retry := input
	retry.Now = now
	retry.LeaseUntil = now.Add(time.Minute)
	retry.OwnerToken = "owner-2"
	retry.ObjectKey = "Contract/record-1/file-1/attempt-2/proof.txt"
	reloaded, retryOwner, err := repository.Accept(t.Context(), retry)
	if err != nil || retryOwner || reloaded.OwnerToken != row.OwnerToken ||
		reloaded.ObjectKey != row.ObjectKey {
		t.Fatalf("expired Accept() = %#v owner=%t err=%v", reloaded, retryOwner, err)
	}
	completeHash := input.RequestHash +
		"|sha256:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
	if err := repository.MarkBlobStored(
		t.Context(), row.ID, row.OwnerToken, input.RequestHash, completeHash,
		"Contract/record-1/file-1/.attempts/prior-owner/proof.txt",
		now.Add(time.Hour),
	); !errors.Is(err, ErrFileUploadOperationLostOwnership) {
		t.Fatalf("rebound MarkBlobStored() error = %v, want lost ownership", err)
	}
	var state, requestHash, objectKey string
	if err := db.QueryRow(`SELECT State, RequestHash, ObjectKey
FROM _System_FileUploadOperation WHERE Id = ?`, row.ID).Scan(
		&state, &requestHash, &objectKey,
	); err != nil {
		t.Fatal(err)
	}
	if state != FileUploadOperationPrepared || requestHash != input.RequestHash ||
		objectKey != row.ObjectKey {
		t.Fatalf(
			"rebound MarkBlobStored mutated row = state %q hash %q key %q",
			state, requestHash, objectKey,
		)
	}
	if err := repository.MarkBlobStored(
		t.Context(), row.ID, row.OwnerToken, input.RequestHash, completeHash,
		row.ObjectKey, now.Add(time.Hour),
	); err != nil {
		t.Fatal(err)
	}
	if err := repository.MarkCompleted(
		t.Context(), row.ID, "stale-owner", completeHash, "stale-object",
		now.Add(24*time.Hour),
	); !errors.Is(err, ErrFileUploadOperationLostOwnership) {
		t.Fatalf("stale MarkCompleted() error = %v, want lost ownership", err)
	}
	if err := repository.MarkCompleted(
		t.Context(), row.ID, row.OwnerToken, completeHash, row.ObjectKey,
		now.Add(24*time.Hour),
	); err != nil {
		t.Fatal(err)
	}

	// A newly constructed repository represents a process restart. The same
	// request must resolve the durable completed row without reacquiring it.
	restarted := NewFileUploadOperationRepository(db)
	row, owner, err = restarted.Accept(t.Context(), input)
	if err != nil || owner || row.State != FileUploadOperationCompleted ||
		row.RequestHash != completeHash {
		t.Fatalf("restart Accept() = %#v owner=%t err=%v", row, owner, err)
	}
	changed := input
	changed.RequestHash = "sha256:cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc"
	if _, _, err := restarted.Accept(t.Context(), changed); !errors.Is(err, ErrFileUploadOperationConflict) {
		t.Fatalf("changed request error = %v, want conflict", err)
	}
}