// Integration tests for Service.RenewLease — the lease-heartbeat primitive that
// keeps a slow-but-alive flow's processing row past the stale-reclaim cutoff so
// stale-reclaim (StaleProcessingSeconds) no longer double-executes it.
//
// They run against a live TiDB (skipped when unreachable, the same as the other
// *_integration tests) and prove the correctness properties the heartbeat relies
// on:
//
//  1. A renewed (heartbeating) in-flight row is NOT reclaimed by a concurrent
//     stale-reclaim: RenewLease bumps LastModifiedDate so the row never crosses
//     the stale cutoff (the SAME LastModifiedDate < now-StaleProcessingSeconds
//     fact ReclaimStaleNext reads — no second lease SSoT).
//  2. A non-renewed (crashed) row IS reclaimed after the cutoff — unchanged from
//     today: RenewLease participates in the existing column, it does not change
//     the crash path.
//  3. RenewLease after a finalize (MarkProcessed) is a NO-OP: the CAS
//     (Status='processing' AND ClaimToken=?) can never un-finalize or revive a
//     terminal row.
//  4. RenewLease with a stale/foreign ClaimToken is a NO-OP: a renewal racing a
//     reclaim that re-leased the row under a fresh token touches nothing.
package connector_events

import (
	"context"
	"errors"
	"testing"
	"time"
)

func TestRenewLease_KeepsHeartbeatingRowPastStaleCutoff_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	if _, err := db.Exec(laneHeadTestDDL); err != nil {
		t.Fatalf("create table: %v", err)
	}
	ctx := context.Background()

	const pack = "pack-hb"
	base := time.Date(2026, 6, 25, 10, 0, 0, 0, time.UTC)
	now := base.Add(100000 * time.Second)
	staleCutoff := now.Add(-StaleProcessingSeconds * time.Second)

	// A processing row whose LastModifiedDate is OLD ENOUGH to be stale (before the
	// cutoff) — without a renewal, ReclaimStaleNext would steal it.
	oldLMD := staleCutoff.Add(-time.Minute)
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorEvent
		(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, RouteVersion,
		 DeliverySequence, EnqueueOrdinal, SourceLaneKey, ClaimedBy, ClaimToken, CreatedDate, LastModifiedDate)
		VALUES (?, 'ns', ?, 'webhook.event', ?, 'processing', 1, 0, 1, ?, 'worker-1', 'tok-live', ?, ?)`,
		"E-live", pack, "ik-live", "lane-live", base, oldLMD); err != nil {
		t.Fatalf("insert processing E-live: %v", err)
	}

	svc := NewService(db, deterministicID("reclaim-token"), func() time.Time { return now })

	// Heartbeat: bump LastModifiedDate to now. This must move the row PAST the stale
	// cutoff (now-StaleProcessingSeconds) so a concurrent stale-reclaim cannot steal it.
	if err := svc.RenewLease(ctx, "E-live", "tok-live"); err != nil {
		t.Fatalf("RenewLease: %v", err)
	}

	// A concurrent stale-reclaim must now find NOTHING reclaimable — the row's lease
	// is fresh (just renewed to `now`), so it is not past the cutoff.
	ev, ok, err := svc.ReclaimStaleNext(ctx, pack, "limp-worker")
	if err != nil {
		t.Fatalf("ReclaimStaleNext: %v", err)
	}
	if ok || ev != nil {
		t.Fatalf("ReclaimStaleNext reclaimed %v (ok=%v) after RenewLease — a heartbeating row must NOT be stolen (double-exec)", ev, ok)
	}

	// Sanity: the lease is still held by the original token (not stolen / re-leased).
	var gotToken, gotStatus string
	if err := db.QueryRowContext(ctx, `SELECT ClaimToken, Status FROM _System_ConnectorEvent WHERE Id = ?`, "E-live").Scan(&gotToken, &gotStatus); err != nil {
		t.Fatalf("read back E-live: %v", err)
	}
	if gotToken != "tok-live" || gotStatus != "processing" {
		t.Fatalf("after RenewLease E-live = (token=%q status=%q), want (tok-live, processing) still held by original worker", gotToken, gotStatus)
	}
}

func TestRenewLease_NonRenewedCrashedRowStillReclaimed_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	if _, err := db.Exec(laneHeadTestDDL); err != nil {
		t.Fatalf("create table: %v", err)
	}
	ctx := context.Background()

	const pack = "pack-crash"
	base := time.Date(2026, 6, 25, 10, 0, 0, 0, time.UTC)
	now := base.Add(100000 * time.Second)
	staleCutoff := now.Add(-StaleProcessingSeconds * time.Second)

	// A processing row from a CRASHED worker: LastModifiedDate is before the cutoff
	// and NOBODY renews it. This must still be reclaimed — RenewLease participates in
	// the existing LastModifiedDate column, it does not change the crash path.
	crashedLMD := staleCutoff.Add(-time.Hour)
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorEvent
		(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, RouteVersion,
		 DeliverySequence, EnqueueOrdinal, SourceLaneKey, ClaimedBy, ClaimToken, CreatedDate, LastModifiedDate)
		VALUES (?, 'ns', ?, 'webhook.event', ?, 'processing', 1, 0, 1, ?, 'dead-worker', 'dead-tok', ?, ?)`,
		"E-crashed", pack, "ik-crashed", "lane-crashed", base, crashedLMD); err != nil {
		t.Fatalf("insert processing E-crashed: %v", err)
	}

	svc := NewService(db, deterministicID("reclaim-token"), func() time.Time { return now })

	// No RenewLease here (the worker crashed). ReclaimStaleNext must reclaim it,
	// exactly as today.
	ev, ok, err := svc.ReclaimStaleNext(ctx, pack, "limp-worker")
	if err != nil {
		t.Fatalf("ReclaimStaleNext: %v", err)
	}
	if !ok || ev == nil || ev.ID != "E-crashed" {
		t.Fatalf("ReclaimStaleNext = %v ok=%v, want it to reclaim the crashed E-crashed row (crash path unchanged)", ev, ok)
	}
}

func TestRenewLease_AfterFinalizeIsNoOp_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	if _, err := db.Exec(laneHeadTestDDL); err != nil {
		t.Fatalf("create table: %v", err)
	}
	ctx := context.Background()

	const pack = "pack-final"
	base := time.Date(2026, 6, 25, 10, 0, 0, 0, time.UTC)
	now := base.Add(100000 * time.Second)

	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorEvent
		(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, RouteVersion,
		 DeliverySequence, EnqueueOrdinal, SourceLaneKey, ClaimedBy, ClaimToken, CreatedDate, LastModifiedDate)
		VALUES (?, 'ns', ?, 'webhook.event', ?, 'processing', 1, 0, 1, ?, 'worker-1', 'tok-final', ?, ?)`,
		"E-final", pack, "ik-final", "lane-final", base, base); err != nil {
		t.Fatalf("insert processing E-final: %v", err)
	}

	svc := NewService(db, deterministicID("unused"), func() time.Time { return now })

	// Finalize the row (terminal success). This clears the processing status.
	if err := svc.MarkProcessed(ctx, "E-final", "tok-final"); err != nil {
		t.Fatalf("MarkProcessed: %v", err)
	}

	// A heartbeat goroutine that fires AFTER finalize must do NOTHING — it can never
	// un-finalize the row. RenewLease returns ErrLeaseLost and changes no row.
	err := svc.RenewLease(ctx, "E-final", "tok-final")
	if !errors.Is(err, ErrLeaseLost) {
		t.Fatalf("RenewLease after finalize = %v, want ErrLeaseLost (must be a no-op, cannot revive a terminal row)", err)
	}

	// The row is STILL processed — RenewLease did not flip it back to processing.
	var gotStatus string
	if err := db.QueryRowContext(ctx, `SELECT Status FROM _System_ConnectorEvent WHERE Id = ?`, "E-final").Scan(&gotStatus); err != nil {
		t.Fatalf("read back E-final: %v", err)
	}
	if gotStatus != "processed" {
		t.Fatalf("after RenewLease-post-finalize E-final.Status = %q, want processed (RenewLease must never un-finalize)", gotStatus)
	}
}

func TestRenewLease_StaleTokenIsNoOp_Integration(t *testing.T) {
	db, cleanup := openTiDB(t)
	defer cleanup()
	if _, err := db.Exec(laneHeadTestDDL); err != nil {
		t.Fatalf("create table: %v", err)
	}
	ctx := context.Background()

	const pack = "pack-stale-tok"
	base := time.Date(2026, 6, 25, 10, 0, 0, 0, time.UTC)
	now := base.Add(100000 * time.Second)

	// The row is currently held by worker-B under token 'tok-current' (e.g. a reclaim
	// already re-leased it). worker-A's heartbeat carries the OLD token 'tok-old'.
	freshLMD := now.Add(-time.Second)
	if _, err := db.ExecContext(ctx, `INSERT INTO _System_ConnectorEvent
		(Id, Namespace, InstalledPackId, EventKind, IdempotencyKey, Status, RouteVersion,
		 DeliverySequence, EnqueueOrdinal, SourceLaneKey, ClaimedBy, ClaimToken, CreatedDate, LastModifiedDate)
		VALUES (?, 'ns', ?, 'webhook.event', ?, 'processing', 1, 0, 1, ?, 'worker-B', 'tok-current', ?, ?)`,
		"E-tok", pack, "ik-tok", "lane-tok", base, freshLMD); err != nil {
		t.Fatalf("insert processing E-tok: %v", err)
	}

	svc := NewService(db, deterministicID("unused"), func() time.Time { return now })

	// worker-A renews with its OLD token — must NOT touch worker-B's lease.
	err := svc.RenewLease(ctx, "E-tok", "tok-old")
	if !errors.Is(err, ErrLeaseLost) {
		t.Fatalf("RenewLease with stale token = %v, want ErrLeaseLost (cannot renew a lease the worker no longer holds)", err)
	}

	// worker-B's lease (token + LastModifiedDate) is unchanged.
	var gotToken string
	var gotLMD time.Time
	if err := db.QueryRowContext(ctx, `SELECT ClaimToken, LastModifiedDate FROM _System_ConnectorEvent WHERE Id = ?`, "E-tok").Scan(&gotToken, &gotLMD); err != nil {
		t.Fatalf("read back E-tok: %v", err)
	}
	if gotToken != "tok-current" {
		t.Fatalf("after stale-token RenewLease E-tok.ClaimToken = %q, want tok-current untouched", gotToken)
	}
	if !gotLMD.Equal(freshLMD) {
		t.Fatalf("after stale-token RenewLease E-tok.LastModifiedDate = %v, want %v untouched (no foreign-token bump)", gotLMD, freshLMD)
	}
}