package entity

import (
	"context"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/tidbcloud/dataflow-service/app/utils"
)

func TestBackup_CRName(t *testing.T) {
	backupID := ID(2988830)
	v1Backup := &Backup{
		BackupSchedule: &BackupSchedule{Name: "dev-tokyo-shared-01-stg-satudora-20260526-before-2"},
		Updatable:      Updatable{Base: Base{ID: backupID}},
		ExtraInfo:      &BackupExtra{BackupCRNameVersion: "v1"},
	}
	assert.Equal(t, utils.UniqueIdentifierLte63("backup-2988830"), v1Backup.CRName())
	assert.LessOrEqual(t, len(v1Backup.CRName()), 63)

	legacyBackup := &Backup{
		BackupSchedule: &BackupSchedule{Name: "my-backup"},
		Updatable:      Updatable{Base: Base{ID: ID(1)}},
	}
	assert.Equal(t, "my-backup-1", legacyBackup.CRName())

	withOld := &Backup{OldBackupName: "legacy-cr-name"}
	assert.Equal(t, "legacy-cr-name", withOld.CRName())
}

func TestLogBackup_CRName(t *testing.T) {
	backupID := ID(2988830)
	v1Backup := &LogBackup{
		Updatable:      Updatable{Base: Base{ID: backupID}},
		BackupSchedule: &BackupSchedule{Name: "dev-tokyo-shared-01-stg-satudora-20260526-before-2"},
		ExtraInfo:      &LogBackupExtra{BackupCRNameVersion: "v1"},
	}
	assert.Equal(t, "log-2988830", v1Backup.CRName())
	assert.LessOrEqual(t, len(v1Backup.CRName()), 63)

	legacyBackup := &LogBackup{
		Updatable:      Updatable{Base: Base{ID: backupID}},
		BackupSchedule: &BackupSchedule{Name: "my-backup"},
	}
	assert.Equal(t, "log-my-backup-2988830", legacyBackup.CRName())
}

func TestCompactBackup_GenerateDownstreamID(t *testing.T) {
	cb := &CompactBackup{
		Updatable: Updatable{Base: Base{ID: ID(42)}},
	}
	assert.Equal(t, "compact-42", cb.GenerateDownstreamID())
}

func TestBackup_GetBucketByRegion(t *testing.T) {
	snapshot := &BucketSetting{
		Buckets: []BackupSettingBucket{
			{BucketId: 101, Regions: []string{"us-east-1", "us-east-2"}},
			{BucketId: 202, Regions: []string{"eu-west-1"}},
		},
	}

	tests := map[string]struct {
		backup     *Backup
		region     string
		wantBucket BackupSettingBucket
		wantFound  bool
	}{
		"matches first bucket primary region": {
			backup:     &Backup{BucketSettingSnapshot: snapshot},
			region:     "us-east-1",
			wantBucket: BackupSettingBucket{BucketId: 101, Regions: []string{"us-east-1", "us-east-2"}},
			wantFound:  true,
		},
		"matches first bucket secondary region": {
			backup:     &Backup{BucketSettingSnapshot: snapshot},
			region:     "us-east-2",
			wantBucket: BackupSettingBucket{BucketId: 101, Regions: []string{"us-east-1", "us-east-2"}},
			wantFound:  true,
		},
		"matches second bucket": {
			backup:     &Backup{BucketSettingSnapshot: snapshot},
			region:     "eu-west-1",
			wantBucket: BackupSettingBucket{BucketId: 202, Regions: []string{"eu-west-1"}},
			wantFound:  true,
		},
		"region not in any bucket": {
			backup:    &Backup{BucketSettingSnapshot: snapshot},
			region:    "ap-northeast-1",
			wantFound: false,
		},
		"nil BucketSettingSnapshot": {
			backup:    &Backup{BucketSettingSnapshot: nil},
			region:    "us-east-1",
			wantFound: false,
		},
		"snapshot with no buckets": {
			backup:    &Backup{BucketSettingSnapshot: &BucketSetting{Buckets: nil}},
			region:    "us-east-1",
			wantFound: false,
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			gotBucket, gotFound := tc.backup.GetBucketByRegion(tc.region)
			assert.Equal(t, tc.wantFound, gotFound)
			assert.Equal(t, tc.wantBucket, gotBucket)
		})
	}
}

// strPtr is a test helper that returns a pointer to a string literal.
func strPtr(s string) *string { return &s }

// tsoStr converts a Unix timestamp (seconds) to the TSO string format used by
// log-backup fields (StartTS, CheckpointTS, TruncateUntil).
func tsoStr(unixSec int64) string { return utils.TransToTSOStr(unixSec) }

func TestLogBackup_ParseStartAndCheckPointAndTruncateUntilTS(t *testing.T) {
	const (
		startUnix      int64 = 1700000000
		checkpointUnix int64 = 1700003600 // 1 h after start
		truncateUnix   int64 = 1700001800 // 30 min after start
	)

	tests := map[string]struct {
		lb           *LogBackup
		wantStart    int64
		wantChk      int64
		wantTruncate int64
		wantErr      bool
	}{
		"nil receiver returns error": {
			lb:      nil,
			wantErr: true,
		},
		"nil StartTS returns error": {
			lb:      &LogBackup{CheckpointTS: tsoStr(checkpointUnix)},
			wantErr: true,
		},
		"empty CheckpointTS returns error": {
			lb:      &LogBackup{StartTS: strPtr(tsoStr(startUnix))},
			wantErr: true,
		},
		"valid: no TruncateUntil returns zero for third value": {
			lb: &LogBackup{
				StartTS:      strPtr(tsoStr(startUnix)),
				CheckpointTS: tsoStr(checkpointUnix),
			},
			wantStart:    startUnix,
			wantChk:      checkpointUnix,
			wantTruncate: 0,
		},
		"valid: empty TruncateUntil treated same as missing": {
			lb: &LogBackup{
				StartTS:       strPtr(tsoStr(startUnix)),
				CheckpointTS:  tsoStr(checkpointUnix),
				TruncateUntil: "",
			},
			wantStart:    startUnix,
			wantChk:      checkpointUnix,
			wantTruncate: 0,
		},
		"valid: TruncateUntil set returns its parsed value": {
			lb: &LogBackup{
				StartTS:       strPtr(tsoStr(startUnix)),
				CheckpointTS:  tsoStr(checkpointUnix),
				TruncateUntil: tsoStr(truncateUnix),
			},
			wantStart:    startUnix,
			wantChk:      checkpointUnix,
			wantTruncate: truncateUnix,
		},
		"invalid StartTS returns error": {
			lb: &LogBackup{
				StartTS:      strPtr("not-a-number"),
				CheckpointTS: tsoStr(checkpointUnix),
			},
			wantErr: true,
		},
		"invalid CheckpointTS returns error": {
			lb: &LogBackup{
				StartTS:      strPtr(tsoStr(startUnix)),
				CheckpointTS: "not-a-number",
			},
			wantErr: true,
		},
		"invalid TruncateUntil returns error": {
			lb: &LogBackup{
				StartTS:       strPtr(tsoStr(startUnix)),
				CheckpointTS:  tsoStr(checkpointUnix),
				TruncateUntil: "not-a-number",
			},
			wantErr: true,
		},
		// CheckpointTSEmpty() now treats "0" as empty — same guard as GenLogBackupValidRange
		"CheckpointTS zero string returns error": {
			lb: &LogBackup{
				StartTS:      strPtr(tsoStr(startUnix)),
				CheckpointTS: "0",
			},
			wantErr: true,
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			gotStart, gotChk, gotTruncate, err := tc.lb.ParseStartAndCheckPointAndTruncateUntilTS()
			if tc.wantErr {
				assert.Error(t, err)
				return
			}
			assert.NoError(t, err)
			assert.Equal(t, tc.wantStart, gotStart)
			assert.Equal(t, tc.wantChk, gotChk)
			assert.Equal(t, tc.wantTruncate, gotTruncate)
		})
	}
}

func TestLogBackup_GenLogBackupValidRange(t *testing.T) {
	const (
		startUnix       int64 = 1700000000
		checkpointUnix  int64 = 1700003600 // 1 h after start
		truncateMidUnix int64 = 1700001800 // 30 min after start, before checkpoint
	)

	tests := map[string]struct {
		lb        *LogBackup
		wantNil   bool // expect (nil, nil, nil) — no valid range, no error
		wantStart int64
		wantEnd   int64
		wantErr   bool
	}{
		// ── guard conditions that return (nil, nil, nil) ─────────────────────
		"nil receiver returns nil range": {
			lb:      nil,
			wantNil: true,
		},
		"nil StartTS returns nil range": {
			lb:      &LogBackup{CheckpointTS: tsoStr(checkpointUnix)},
			wantNil: true,
		},
		"empty CheckpointTS returns nil range": {
			lb:      &LogBackup{StartTS: strPtr(tsoStr(startUnix))},
			wantNil: true,
		},

		// ── valid ranges ──────────────────────────────────────────────────────
		"no TruncateUntil: returns [start, checkpoint)": {
			lb: &LogBackup{
				StartTS:      strPtr(tsoStr(startUnix)),
				CheckpointTS: tsoStr(checkpointUnix),
			},
			wantStart: startUnix,
			wantEnd:   checkpointUnix,
		},
		"TruncateUntil before start: effective start unchanged": {
			lb: &LogBackup{
				StartTS:       strPtr(tsoStr(startUnix)),
				CheckpointTS:  tsoStr(checkpointUnix),
				TruncateUntil: tsoStr(startUnix - 1),
			},
			wantStart: startUnix,
			wantEnd:   checkpointUnix,
		},
		"TruncateUntil equals start: effective start unchanged": {
			// condition is truncateUntil > logBackupStartTS (strict), so equal does not advance
			lb: &LogBackup{
				StartTS:       strPtr(tsoStr(startUnix)),
				CheckpointTS:  tsoStr(checkpointUnix),
				TruncateUntil: tsoStr(startUnix),
			},
			wantStart: startUnix,
			wantEnd:   checkpointUnix,
		},
		"TruncateUntil between start and checkpoint: advances effective start": {
			lb: &LogBackup{
				StartTS:       strPtr(tsoStr(startUnix)),
				CheckpointTS:  tsoStr(checkpointUnix),
				TruncateUntil: tsoStr(truncateMidUnix),
			},
			wantStart: truncateMidUnix,
			wantEnd:   checkpointUnix,
		},

		// ── empty / exhausted ranges ──────────────────────────────────────────
		"startTS equals checkpointTS: returns nil range": {
			lb: &LogBackup{
				StartTS:      strPtr(tsoStr(checkpointUnix)),
				CheckpointTS: tsoStr(checkpointUnix),
			},
			wantNil: true,
		},
		"startTS exceeds checkpointTS: returns nil range": {
			lb: &LogBackup{
				StartTS:      strPtr(tsoStr(checkpointUnix)),
				CheckpointTS: tsoStr(startUnix), // reversed: checkpoint is earlier
			},
			wantNil: true,
		},
		"TruncateUntil equals checkpoint: effective start == checkpoint, nil range": {
			lb: &LogBackup{
				StartTS:       strPtr(tsoStr(startUnix)),
				CheckpointTS:  tsoStr(checkpointUnix),
				TruncateUntil: tsoStr(checkpointUnix),
			},
			wantNil: true,
		},
		"TruncateUntil exceeds checkpoint: nil range": {
			lb: &LogBackup{
				StartTS:       strPtr(tsoStr(startUnix)),
				CheckpointTS:  tsoStr(checkpointUnix),
				TruncateUntil: tsoStr(checkpointUnix + 3600),
			},
			wantNil: true,
		},
		// CheckpointTS "0" is caught by CheckpointTSEmpty() guard — returns nil before any parsing
		"CheckpointTS zero string: returns nil range": {
			lb: &LogBackup{
				StartTS:      strPtr(tsoStr(startUnix)),
				CheckpointTS: "0",
			},
			wantNil: true,
		},

		// ── parse errors propagated ───────────────────────────────────────────
		"invalid StartTS: returns error": {
			lb: &LogBackup{
				StartTS:      strPtr("not-a-number"),
				CheckpointTS: tsoStr(checkpointUnix),
			},
			wantErr: true,
		},
		"invalid CheckpointTS: returns error": {
			lb: &LogBackup{
				StartTS:      strPtr(tsoStr(startUnix)),
				CheckpointTS: "not-a-number",
			},
			wantErr: true,
		},
		"invalid TruncateUntil: returns error": {
			lb: &LogBackup{
				StartTS:       strPtr(tsoStr(startUnix)),
				CheckpointTS:  tsoStr(checkpointUnix),
				TruncateUntil: "not-a-number",
			},
			wantErr: true,
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			gotStart, gotEnd, err := tc.lb.GenLogBackupValidRange()
			if tc.wantErr {
				assert.Error(t, err)
				assert.Nil(t, gotStart)
				assert.Nil(t, gotEnd)
				return
			}
			assert.NoError(t, err)
			if tc.wantNil {
				assert.Nil(t, gotStart)
				assert.Nil(t, gotEnd)
			} else {
				if assert.NotNil(t, gotStart) {
					assert.Equal(t, tc.wantStart, *gotStart)
				}
				if assert.NotNil(t, gotEnd) {
					assert.Equal(t, tc.wantEnd, *gotEnd)
				}
			}
		})
	}
}

func TestLogBackup_GenPiTRTimeRanges(t *testing.T) {
	const (
		logStartUnix   int64 = 1700000000
		checkpointUnix int64 = 1700007200
		bucketID       ID    = 101
	)

	makeLogBackup := func(startUnix, chkUnix int64, bktID ID) *LogBackup {
		s := tsoStr(startUnix)
		return &LogBackup{
			StartTS:      &s,
			CheckpointTS: tsoStr(chkUnix),
			BucketID:     bktID,
		}
	}
	makeFullBackup := func(startUnix int64, bktID ID) Backup {
		return Backup{
			Updatable:      Updatable{Base: Base{ID: ID(startUnix)}},
			Status:         BackupStatusSuccess,
			StartTimeStamp: tsoStr(startUnix),
			BucketID:       bktID,
		}
	}

	tests := map[string]struct {
		lb         *LogBackup
		backups    []Backup
		wantRanges []struct {
			startTS  int64
			endTS    int64
			backupTS int64
		}
	}{
		"invalid log backup range returns nil": {
			lb:      &LogBackup{},
			backups: []Backup{makeFullBackup(logStartUnix+1800, bucketID)},
		},
		"returns eligible ranges in input order": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				makeFullBackup(logStartUnix+5400, bucketID),
				{Status: "failed", StartTimeStamp: tsoStr(logStartUnix), BucketID: bucketID},
				makeFullBackup(logStartUnix+3600, bucketID+1),
				makeFullBackup(logStartUnix+1800, bucketID),
				makeFullBackup(logStartUnix-1, bucketID),    // break
				makeFullBackup(logStartUnix+1200, bucketID), // unreachable due to break
			},
			wantRanges: []struct {
				startTS  int64
				endTS    int64
				backupTS int64
			}{
				{startTS: logStartUnix + 5400 + 1, endTS: checkpointUnix, backupTS: logStartUnix + 5400},
				{startTS: logStartUnix + 1800 + 1, endTS: checkpointUnix, backupTS: logStartUnix + 1800},
			},
		},
		"encountering too-old backup stops scanning remaining backups": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				makeFullBackup(logStartUnix+3600, bucketID),
				makeFullBackup(logStartUnix-1, bucketID),    // break
				makeFullBackup(logStartUnix+1800, bucketID), // unreachable due to break
			},
			wantRanges: []struct {
				startTS  int64
				endTS    int64
				backupTS int64
			}{
				{startTS: logStartUnix + 3600 + 1, endTS: checkpointUnix, backupTS: logStartUnix + 3600},
			},
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			got := tc.lb.GenPiTRTimeRanges(context.Background(), tc.backups)
			assert.Len(t, got, len(tc.wantRanges))
			for i, want := range tc.wantRanges {
				if i >= len(got) {
					break
				}
				assert.Equal(t, want.startTS, got[i].StartTS)
				assert.Equal(t, want.endTS, got[i].EndTS)
				assert.Equal(t, tsoStr(want.backupTS), got[i].Backup.StartTimeStamp)
				assert.Equal(t, bucketID, got[i].LogBackup.BucketID)
			}
		})
	}
}

func TestLogBackup_GenPiTRTimeRange(t *testing.T) {
	const (
		logStartUnix   int64 = 1700000000
		checkpointUnix int64 = 1700007200 // 2 h after log start
		bucketID       ID    = 101
	)

	makeLogBackup := func(startUnix, chkUnix int64, bktID ID) *LogBackup {
		s := tsoStr(startUnix)
		return &LogBackup{
			StartTS:      &s,
			CheckpointTS: tsoStr(chkUnix),
			BucketID:     bktID,
		}
	}
	makeFullBackup := func(startUnix int64, bktID ID) Backup {
		return Backup{
			Status:         BackupStatusSuccess,
			StartTimeStamp: tsoStr(startUnix),
			BucketID:       bktID,
			// TableFilter nil → IsFullClusterBackup() == true
		}
	}

	tests := map[string]struct {
		lb        *LogBackup
		backups   []Backup
		wantFound bool
		wantStart int64
		wantEnd   int64
	}{
		// ── no valid log backup range ──────────────────────────────────────────
		"nil log backup: not found": {
			lb:        nil,
			wantFound: false,
		},
		"no StartTS or CheckpointTS: not found": {
			lb:        &LogBackup{},
			wantFound: false,
		},

		// ── no valid full backups ──────────────────────────────────────────────
		"empty backups list: not found": {
			lb:        makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			wantFound: false,
		},
		"backup with non-success status: not found": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				{Status: "failed", StartTimeStamp: tsoStr(logStartUnix + 1800), BucketID: bucketID},
			},
			wantFound: false,
		},
		"partial backup (has table filter): not found": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				{
					Status:         BackupStatusSuccess,
					StartTimeStamp: tsoStr(logStartUnix + 1800),
					BucketID:       bucketID,
					TableFilter:    TableFilter{"db1.table1"},
				},
			},
			wantFound: false,
		},
		"backup from different bucket: not found": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				makeFullBackup(logStartUnix+1800, bucketID+1),
			},
			wantFound: false,
		},
		"backup before log start (backupTS < startTS): not found": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				makeFullBackup(logStartUnix-3600, bucketID),
			},
			wantFound: false,
		},
		"backup at checkpoint (backupTS == endTS): sentinel stays, not found": {
			// left starts at checkpointTS; backupTS == checkpointTS fails backupTS < left, so left stays at sentinel → not found
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				makeFullBackup(checkpointUnix, bucketID),
			},
			wantFound: false,
		},
		"backup after checkpoint: not found": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				makeFullBackup(checkpointUnix+3600, bucketID),
			},
			wantFound: false,
		},
		"backup with invalid StartTimeStamp: skipped, not found": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				{Status: BackupStatusSuccess, StartTimeStamp: "not-a-tso", BucketID: bucketID},
			},
			wantFound: false,
		},

		// ── valid full backups found ───────────────────────────────────────────
		"backup within range: found, startTS advanced to backupTS+1": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				makeFullBackup(logStartUnix+1800, bucketID),
			},
			wantFound: true,
			wantStart: logStartUnix + 1800 + 1,
			wantEnd:   checkpointUnix,
		},
		"backup at log start (backupTS == startTS): found": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				makeFullBackup(logStartUnix, bucketID),
			},
			wantFound: true,
			wantStart: logStartUnix + 1,
			wantEnd:   checkpointUnix,
		},
		// Backups sorted newest-first (descending). The loop does NOT break early —
		// it scans all in-range backups and keeps updating left = min(left, backupTS).
		// So the OLDEST qualifying backup sets the final left boundary.
		"multiple backups descending: picks oldest (smallest backupTS)": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				makeFullBackup(logStartUnix+3600, bucketID), // newest
				makeFullBackup(logStartUnix+1800, bucketID), // older → becomes the final left
			},
			wantFound: true,
			wantStart: logStartUnix + 1800 + 1, // oldest backup + 1
			wantEnd:   checkpointUnix,
		},
		// Three backups all in-range: algorithm scans all, left ends up at the oldest.
		"three backups all in range: oldest wins": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				makeFullBackup(logStartUnix+5400, bucketID), // 1.5 h in
				makeFullBackup(logStartUnix+3600, bucketID), // 1 h in
				makeFullBackup(logStartUnix+1800, bucketID), // 30 min in → final left
			},
			wantFound: true,
			wantStart: logStartUnix + 1800 + 1,
			wantEnd:   checkpointUnix,
		},
		// In-range backups followed by a too-old backup that triggers the break.
		// The break does not discard already-found backups.
		"backups spanning break boundary: in-range backups found before break": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				makeFullBackup(logStartUnix+3600, bucketID), // in range
				makeFullBackup(logStartUnix+1800, bucketID), // in range → final left
				makeFullBackup(logStartUnix-1, bucketID),    // backupTS < startTS → break
			},
			wantFound: true,
			wantStart: logStartUnix + 1800 + 1,
			wantEnd:   checkpointUnix,
		},
		// Backup just one second before checkpoint: left = checkpointTS-1, left+1 = checkpointTS.
		// A degenerate but valid range where left == right.
		"backup just below checkpoint: degenerate range left==right": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				makeFullBackup(checkpointUnix-1, bucketID),
			},
			wantFound: true,
			wantStart: checkpointUnix, // (checkpointUnix-1) + 1
			wantEnd:   checkpointUnix,
		},
		// Backup that comes right at startTS triggers the break when it IS below startTS.
		// Verify: only backup at (startTS-1) → immediate break → not found.
		"only backup is one second before startTS: breaks immediately, not found": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				makeFullBackup(logStartUnix-1, bucketID),
			},
			wantFound: false,
		},
		"invalid backup skipped, valid backup after: found": {
			lb: makeLogBackup(logStartUnix, checkpointUnix, bucketID),
			backups: []Backup{
				{Status: "failed", StartTimeStamp: tsoStr(logStartUnix + 3600), BucketID: bucketID},
				makeFullBackup(logStartUnix+1800, bucketID),
			},
			wantFound: true,
			wantStart: logStartUnix + 1800 + 1,
			wantEnd:   checkpointUnix,
		},
	}

	for name, tc := range tests {
		t.Run(name, func(t *testing.T) {
			gotStart, gotEnd, gotFound := tc.lb.GenPiTRTimeRange(context.Background(), tc.backups)
			assert.Equal(t, tc.wantFound, gotFound)
			if tc.wantFound {
				assert.Equal(t, tc.wantStart, gotStart)
				assert.Equal(t, tc.wantEnd, gotEnd)
			} else {
				assert.Equal(t, int64(0), gotStart)
				assert.Equal(t, int64(0), gotEnd)
			}
		})
	}
}