package db

import (
	"testing"

	"tangled.org/core/notifier"
)

func TestMillLeaseRoundTrip(t *testing.T) {
	d := newTestDB(t)

	lease := MillLease{
		LeaseID:  "lease-1",
		NodeID:   "node-1",
		Engine:   "dummy",
		Knot:     "knot.example",
		Rkey:     "rkey1",
		Workflow: "build",
		State:    "reserved",
	}
	if err := d.SaveMillLease(lease); err != nil {
		t.Fatalf("SaveMillLease: %v", err)
	}

	lease.State = "running"
	if err := d.SaveMillLease(lease); err != nil {
		t.Fatalf("SaveMillLease(transition): %v", err)
	}

	leases, err := d.ListMillLeases()
	if err != nil {
		t.Fatalf("ListMillLeases: %v", err)
	}
	if len(leases) != 1 {
		t.Fatalf("ListMillLeases returned %d leases, want 1 (state transition must replace, not duplicate)", len(leases))
	}
	if leases[0] != lease {
		t.Fatalf("ListMillLeases[0] = %+v, want %+v", leases[0], lease)
	}

	if err := d.DeleteMillLease("lease-1"); err != nil {
		t.Fatalf("DeleteMillLease: %v", err)
	}
	if leases, _ = d.ListMillLeases(); len(leases) != 0 {
		t.Fatalf("lease survived deletion: %+v", leases)
	}
}

func TestDeleteMillLeasesByNode(t *testing.T) {
	d := newTestDB(t)

	for _, l := range []MillLease{
		{LeaseID: "a", NodeID: "node-1", Engine: "dummy", Knot: "k", Rkey: "r1", Workflow: "w", State: "running"},
		{LeaseID: "b", NodeID: "node-1", Engine: "dummy", Knot: "k", Rkey: "r2", Workflow: "w", State: "reserved"},
		{LeaseID: "c", NodeID: "node-2", Engine: "dummy", Knot: "k", Rkey: "r3", Workflow: "w", State: "running"},
	} {
		if err := d.SaveMillLease(l); err != nil {
			t.Fatalf("SaveMillLease(%s): %v", l.LeaseID, err)
		}
	}

	if err := d.DeleteMillLeasesByNode("node-1"); err != nil {
		t.Fatalf("DeleteMillLeasesByNode: %v", err)
	}
	leases, err := d.ListMillLeases()
	if err != nil {
		t.Fatalf("ListMillLeases: %v", err)
	}
	if len(leases) != 1 || leases[0].LeaseID != "c" {
		t.Fatalf("ListMillLeases = %+v, want only node-2's lease c", leases)
	}
}

func TestExecutorCursors(t *testing.T) {
	d := newTestDB(t)

	if err := d.SetExecutorCursor("node-1", 5); err != nil {
		t.Fatalf("SetExecutorCursor: %v", err)
	}
	if err := d.SetExecutorCursor("node-1", 9); err != nil {
		t.Fatalf("SetExecutorCursor(advance): %v", err)
	}
	if err := d.SetExecutorCursor("node-2", 1); err != nil {
		t.Fatalf("SetExecutorCursor(node-2): %v", err)
	}

	cursors, err := d.ListExecutorCursors()
	if err != nil {
		t.Fatalf("ListExecutorCursors: %v", err)
	}
	if len(cursors) != 2 || cursors["node-1"] != 9 || cursors["node-2"] != 1 {
		t.Fatalf("ListExecutorCursors = %v, want node-1:9 node-2:1", cursors)
	}

	if err := d.DeleteExecutorCursor("node-1"); err != nil {
		t.Fatalf("DeleteExecutorCursor: %v", err)
	}
	if cursors, _ = d.ListExecutorCursors(); len(cursors) != 1 || cursors["node-2"] != 1 {
		t.Fatalf("ListExecutorCursors after delete = %v, want only node-2:1", cursors)
	}
}

func TestCompleteOrphanMillLeaseIsAtomic(t *testing.T) {
	d := newTestDB(t)
	lease := MillLease{
		LeaseID: "lease-1", NodeID: "node-1", Engine: "dummy",
		Knot: "knot.example", Rkey: "rkey1", Workflow: "build", State: "running",
	}
	if err := d.SaveMillLease(lease); err != nil {
		t.Fatalf("SaveMillLease: %v", err)
	}
	if _, err := d.Exec(`
		create trigger reject_mill_lease_delete
		before delete on mill_leases
		begin
			select raise(abort, 'forced delete failure');
		end
	`); err != nil {
		t.Fatalf("create failure trigger: %v", err)
	}

	n := notifier.New()
	notifications := n.Subscribe()
	defer n.Unsubscribe(notifications)
	err := d.CompleteOrphanMillLease(
		"lease-1",
		"at://knot.example/sh.tangled.pipeline/rkey1",
		"build",
		"failed",
		nil,
		nil,
		&n,
	)
	if err == nil {
		t.Fatal("CompleteOrphanMillLease succeeded despite forced lease deletion failure")
	}
	var eventCount int
	if err := d.QueryRow(`select count(*) from events`).Scan(&eventCount); err != nil {
		t.Fatalf("count events after rollback: %v", err)
	}
	if eventCount != 0 {
		t.Fatalf("terminal event count after rollback = %d, want 0", eventCount)
	}
	if leases, listErr := d.ListMillLeases(); listErr != nil || len(leases) != 1 {
		t.Fatalf("leases after rollback = %+v, err = %v; want original lease", leases, listErr)
	}
	select {
	case <-notifications:
		t.Fatal("rollback notified event subscribers")
	default:
	}

	if _, err := d.Exec(`drop trigger reject_mill_lease_delete`); err != nil {
		t.Fatalf("drop failure trigger: %v", err)
	}
	if err := d.CompleteOrphanMillLease(
		"lease-1",
		"at://knot.example/sh.tangled.pipeline/rkey1",
		"build",
		"failed",
		nil,
		nil,
		&n,
	); err != nil {
		t.Fatalf("CompleteOrphanMillLease retry: %v", err)
	}
	if err := d.QueryRow(`select count(*) from events`).Scan(&eventCount); err != nil {
		t.Fatalf("count committed events: %v", err)
	}
	if eventCount != 1 {
		t.Fatalf("terminal event count after commit = %d, want 1", eventCount)
	}
	if leases, listErr := d.ListMillLeases(); listErr != nil || len(leases) != 0 {
		t.Fatalf("leases after commit = %+v, err = %v; want none", leases, listErr)
	}
	select {
	case <-notifications:
	default:
		t.Fatal("committed terminal event did not notify subscribers")
	}
}
