cloud-ip-validator/internal/db/queries_runs_test.go

package db

import (
	"context"
	"database/sql"
	"errors"
	"path/filepath"
	"testing"
	"time"
)

func submit(t *testing.T, d *DB, kind string, addrs ...string) {
	t.Helper()
	if _, err := d.SubmitIPsAs(context.Background(), addrs, kind); err != nil {
		t.Fatal(err)
	}
}

func finish(t *testing.T, d *DB, addr, verdict string, expected int) {
	t.Helper()
	ctx := context.Background()
	ip, err := d.GetIPByAddress(ctx, addr)
	if err != nil {
		t.Fatal(err)
	}
	if err := d.FinishIPExpected(ctx, ip.ID, verdict, expected); err != nil {
		t.Fatal(err)
	}
}

func runs(t *testing.T, d *DB) []RunSummary {
	t.Helper()
	r, err := d.ListRuns(context.Background())
	if err != nil {
		t.Fatal(err)
	}
	return r
}

// An address entering an idle queue opens a run; everything submitted while it
// is open joins it; it is finalized when the last address is done.
func TestRunOpensJoinsAndFinalizes(t *testing.T) {
	d, ctx := newTestDB(t)
	submit(t, d, RunAuto, "1.1.1.1", "2.2.2.2")
	submit(t, d, RunManual, "3.3.3.3") // joins the open run, kind stays auto

	rs := runs(t, d)
	if len(rs) != 1 || rs[0].State != RunOpen || rs[0].Kind != RunAuto || rs[0].Total != 3 || rs[0].Pending != 3 {
		t.Fatalf("expected one open auto run with 3 pending rows: %+v", rs)
	}
	a, _ := d.GetIPByAddress(ctx, "1.1.1.1")
	c, _ := d.GetIPByAddress(ctx, "3.3.3.3")
	if a.ID == 0 || c.ID == 0 {
		t.Fatal("rows missing")
	}

	finish(t, d, "1.1.1.1", ResultPass, 22)
	finish(t, d, "2.2.2.2", ResultPartial, 22)
	if rs = runs(t, d); rs[0].State != RunOpen || rs[0].Addresses != 2 || rs[0].Pending != 1 {
		t.Fatalf("one address still pending, the run stays open: %+v", rs[0])
	}
	finish(t, d, "3.3.3.3", ResultFail, -1)

	rs = runs(t, d)
	if rs[0].State != RunFinalized || rs[0].FinalizedAt == nil || rs[0].Pass != 1 || rs[0].Partial != 1 || rs[0].Fail != 1 || rs[0].Addresses != 3 {
		t.Fatalf("expected a finalized run with the three verdicts: %+v", rs[0])
	}
	res, err := d.ListRunResults(ctx, rs[0].ID)
	if err != nil || len(res) != 3 {
		t.Fatalf("results: %v %v", res, err)
	}
	byIP := map[string]RunResult{}
	for _, r := range res {
		byIP[r.IPAddress] = r
	}
	if byIP["1.1.1.1"].ExpectedChecks != 22 || byIP["3.3.3.3"].ExpectedChecks != -1 || byIP["2.2.2.2"].Verdict != ResultPartial {
		t.Fatalf("results: %+v", byIP)
	}
}

// A re-check after the run is finalized opens a new run and leaves the old one
// as it was; the old cycle's checks keep their run.
func TestRecheckAfterFinalizeOpensNewRun(t *testing.T) {
	d, ctx := newTestDB(t)
	ip := checkingIP(t, d, "1.1.1.1")
	if _, err := d.UpsertCheckIfOpen(ctx, checkOf(ip, "icmp", true)); err != nil {
		t.Fatal(err)
	}
	finish(t, d, "1.1.1.1", ResultPass, 1)
	first := runs(t, d)[0]
	if first.State != RunFinalized {
		t.Fatalf("expected finalized: %+v", first)
	}

	submit(t, d, RunManual, "1.1.1.1") // re-check of a finished address
	rs := runs(t, d)
	if len(rs) != 2 || rs[0].State != RunOpen || rs[0].ID == first.ID {
		t.Fatalf("a re-check after the run ended must open a new run: %+v", rs)
	}
	ip, _ = d.GetIPByAddress(ctx, "1.1.1.1")
	if err := d.SetChecking(ctx, ip.ID, time.Minute); err != nil {
		t.Fatal(err)
	}
	ip, _ = d.GetIP(ctx, ip.ID)
	if _, err := d.UpsertCheckIfOpen(ctx, checkOf(ip, "icmp", false)); err != nil {
		t.Fatal(err)
	}
	finish(t, d, "1.1.1.1", ResultFail, 1)

	var n1, n2 int
	d.QueryRowContext(ctx, `SELECT COUNT(*) FROM checks WHERE run_id=?`, first.ID).Scan(&n1)
	d.QueryRowContext(ctx, `SELECT COUNT(*) FROM checks WHERE run_id=?`, rs[0].ID).Scan(&n2)
	if n1 != 1 || n2 != 1 {
		t.Fatalf("each run keeps its own cycle's check: %d %d", n1, n2)
	}
	old, _ := d.ListRunResults(ctx, first.ID)
	cur, _ := d.ListRunResults(ctx, rs[0].ID)
	if len(old) != 1 || old[0].Verdict != ResultPass || old[0].CycleID != 1 || len(cur) != 1 || cur[0].Verdict != ResultFail || cur[0].CycleID != 2 {
		t.Fatalf("results must not cross: old=%+v cur=%+v", old, cur)
	}
	// The checks of a run are the ones of its result cycle, nothing else.
	var got []RunCheck
	if err := d.EachRunCheck(ctx, first.ID, func(c RunCheck) { got = append(got, c) }); err != nil || len(got) != 1 || !got[0].Success {
		t.Fatalf("run 1 checks: %+v %v", got, err)
	}
}

// A re-check while the run is still open joins it and replaces the address's
// result, so a run holds one result per address.
func TestRecheckInsideOpenRunReplacesResult(t *testing.T) {
	d, ctx := newTestDB(t)
	submit(t, d, RunManual, "1.1.1.1", "2.2.2.2")
	finish(t, d, "1.1.1.1", ResultFail, 1)
	submit(t, d, RunManual, "1.1.1.1") // while 2.2.2.2 is still pending
	finish(t, d, "1.1.1.1", ResultPass, 1)
	finish(t, d, "2.2.2.2", ResultPass, 1)

	rs := runs(t, d)
	if len(rs) != 1 || rs[0].Addresses != 2 || rs[0].Pass != 2 || rs[0].Fail != 0 {
		t.Fatalf("expected one run with the latest verdicts: %+v", rs)
	}
	res, _ := d.ListRunResults(ctx, rs[0].ID)
	for _, r := range res {
		if r.IPAddress == "1.1.1.1" && r.CycleID != 2 {
			t.Fatalf("the re-checked address must show its latest cycle: %+v", r)
		}
	}
	if n, err := d.CountRecheckedInRun(ctx, rs[0].ID); err != nil || n != 0 {
		t.Fatalf("no checks stored, so no re-check counted: %d %v", n, err)
	}
}

func TestRunEndsWhenQueueIsClearedOrDeleted(t *testing.T) {
	d, ctx := newTestDB(t)
	submit(t, d, RunManual, "1.1.1.1", "2.2.2.2")
	finish(t, d, "1.1.1.1", ResultPass, 1)
	if _, err := d.ClearAllIPs(ctx); err != nil {
		t.Fatal(err)
	}
	rs := runs(t, d)
	if len(rs) != 1 || rs[0].State != RunFinalized || rs[0].Addresses != 1 {
		t.Fatalf("clearing ends the run with what it has: %+v", rs)
	}

	// The next submission is a new run, not a join of the ended one.
	submit(t, d, RunManual, "3.3.3.3")
	if rs = runs(t, d); len(rs) != 2 || rs[0].State != RunOpen {
		t.Fatalf("expected a new open run: %+v", rs)
	}
	// A run with no result at all disappears when its rows are deleted.
	ip, _ := d.GetIPByAddress(ctx, "3.3.3.3")
	if err := d.DeleteIP(ctx, ip.ID); err != nil {
		t.Fatal(err)
	}
	if rs = runs(t, d); len(rs) != 1 {
		t.Fatalf("an empty run must be dropped: %+v", rs)
	}
}

func TestCancelAndRetryFailureRecordResults(t *testing.T) {
	d, ctx := newTestDB(t)
	submit(t, d, RunManual, "1.1.1.1", "2.2.2.2")
	a, _ := d.GetIPByAddress(ctx, "1.1.1.1")
	if err := d.CancelIP(ctx, a.ID); err != nil {
		t.Fatal(err)
	}
	b, _ := d.GetIPByAddress(ctx, "2.2.2.2")
	if err := d.RequeueOrFail(ctx, b.ID, "", 0); err != nil { // retries exhausted at once
		t.Fatal(err)
	}
	rs := runs(t, d)
	if rs[0].State != RunFinalized || rs[0].Cancelled != 1 || rs[0].Fail != 1 {
		t.Fatalf("cancelled and failed addresses are results of the run: %+v", rs[0])
	}
}

// Ingress checks name the validator that held the address.
func TestIngressCheckTakesValidatorOfTheAddress(t *testing.T) {
	d, ctx := newTestDB(t)
	if err := d.RegisterValidator(ctx, "validator-7", "host", "port", "v"); err != nil {
		t.Fatal(err)
	}
	ip := checkingIP(t, d, "1.1.1.1")
	if _, err := d.ExecContext(ctx, `UPDATE ip_queue SET owner_validator_id='validator-7' WHERE id=?`, ip.ID); err != nil {
		t.Fatal(err)
	}
	if _, err := d.UpsertCheckIfOpen(ctx, checkOf(ip, "icmp", true)); err != nil {
		t.Fatal(err)
	}
	var v string
	if err := d.QueryRowContext(ctx, `SELECT validator_id FROM checks WHERE ip_id=?`, ip.ID).Scan(&v); err != nil || v != "validator-7" {
		t.Fatalf("validator of an ingress check = %q err=%v", v, err)
	}
	// An explicit validator (egress checks) is kept.
	c := checkOf(ip, "https", true)
	c.Source, c.ValidatorID, c.Target = SourceEgress, "validator-9", "https://x"
	if _, err := d.UpsertCheckIfOpen(ctx, c); err != nil {
		t.Fatal(err)
	}
	if err := d.QueryRowContext(ctx, `SELECT validator_id FROM checks WHERE ip_id=? AND source='egress'`, ip.ID).Scan(&v); err != nil || v != "validator-9" {
		t.Fatalf("egress validator = %q err=%v", v, err)
	}
}

func TestSubnetsReplaceAndValidate(t *testing.T) {
	d, ctx := newTestDB(t)
	if err := d.ReplaceSubnets(ctx, []Subnet{{CIDR: "10.1.2.3/24", Label: "a"}, {CIDR: "10.0.0.0/8"}, {CIDR: "10.1.2.0/24", Label: "dup"}}); err != nil {
		t.Fatal(err)
	}
	got, err := d.ListSubnets(ctx)
	if err != nil || len(got) != 2 || got[0].CIDR != "10.0.0.0/8" || got[1].CIDR != "10.1.2.0/24" {
		t.Fatalf("subnets must be canonical, de-duplicated and sorted by prefix: %+v %v", got, err)
	}
	if err := d.ReplaceSubnets(ctx, []Subnet{{CIDR: "nonsense"}}); !errors.Is(err, ErrValidation) {
		t.Fatalf("expected ErrValidation, got %v", err)
	}
	if got, _ := d.ListSubnets(ctx); len(got) != 2 {
		t.Fatalf("a rejected list must leave the old one: %+v", got)
	}
	if err := d.ReplaceSubnets(ctx, nil); err != nil {
		t.Fatal(err)
	}
	if got, _ := d.ListSubnets(ctx); len(got) != 0 {
		t.Fatalf("an empty list clears: %+v", got)
	}
}

func TestRegistryFilterByRunAndSubnet(t *testing.T) {
	d, ctx := newTestDB(t)
	submit(t, d, RunManual, "10.0.0.1", "10.0.0.2", "10.0.1.1")
	for _, a := range []string{"10.0.0.1", "10.0.0.2", "10.0.1.1"} {
		finish(t, d, a, ResultPass, -1)
	}
	runID := runs(t, d)[0].ID
	submit(t, d, RunManual, "10.0.0.2") // second run holds only this address
	finish(t, d, "10.0.0.2", ResultFail, -1)
	secondID := runs(t, d)[0].ID

	count := func(f RegistryFilter) int {
		t.Helper()
		_, total, err := d.ListRegistryPage(ctx, f, 50, 0)
		if err != nil {
			t.Fatal(err)
		}
		return total
	}
	if n := count(RegistryFilter{RunID: runID}); n != 3 {
		t.Errorf("run 1: %d", n)
	}
	if n := count(RegistryFilter{RunID: secondID}); n != 1 {
		t.Errorf("run 2: %d", n)
	}
	if n := count(RegistryFilter{Subnet: "10.0.0.0/24"}); n != 2 {
		t.Errorf("subnet /24: %d", n)
	}
	if n := count(RegistryFilter{RunID: secondID, Subnet: "10.0.1.0/24"}); n != 0 {
		t.Errorf("run 2 and the other subnet: %d", n)
	}
	if n := count(RegistryFilter{Subnet: "192.168.0.0/16"}); n != 0 {
		t.Errorf("subnet with no address: %d", n)
	}
	if _, _, err := d.ListRegistryPage(ctx, RegistryFilter{Subnet: "x"}, 10, 0); !errors.Is(err, ErrValidation) {
		t.Errorf("bad subnet: %v", err)
	}
}

// Migration 0011 on a database of version 10: runs are cut at pauses of more
// than an hour, results come from the queue row or the checks, ingress checks
// get their validator from the fip_associated event, and live queue rows
// without a run are adopted into an open run when the database is opened.
func TestMigration0011BuildsRunsFromExistingData(t *testing.T) {
	ctx := context.Background()
	path := filepath.Join(t.TempDir(), "old.db")
	raw, err := sql.Open("sqlite", path)
	if err != nil {
		t.Fatal(err)
	}
	raw.SetMaxOpenConns(1)
	for _, m := range migrations {
		if m.version > 10 {
			break
		}
		if _, err := raw.ExecContext(ctx, m.sql); err != nil {
			t.Fatalf("migration %d: %v", m.version, err)
		}
	}
	raw.ExecContext(ctx, `PRAGMA user_version=10`)
	exec := func(q string, args ...any) {
		t.Helper()
		if _, err := raw.ExecContext(ctx, q, args...); err != nil {
			t.Fatalf("seed: %v\n%s", err, q)
		}
	}
	const ts = "2026-10-02T13:00:00Z"
	for i, ip := range []string{"1.1.1.1", "2.2.2.2", "3.3.3.3"} {
		exec(`INSERT INTO ip_registry (id, ip_address, first_seen_at, last_seen_at, next_cycle, created_at, updated_at) VALUES (?, ?, ?, ?, 3, ?, ?)`, i+1, ip, ts, ts, ts, ts)
	}
	// 1.1.1.1 and 2.2.2.2 finish minutes apart (run 1); 1.1.1.1 is checked
	// again three hours later (run 2). 3.3.3.3 is still in the queue.
	exec(`INSERT INTO ip_queue (id, ip_address, sequence, state, overall_result, aggregated_at, registry_id, cycle_id, created_at, updated_at)
		VALUES (1, '1.1.1.1', 1, 'done', 'fail', '2026-10-02T16:00:30Z', 1, 2, ?, ?), (2, '2.2.2.2', 2, 'done', 'pass', '2026-10-02T13:05:30Z', 2, 1, ?, ?),
		       (3, '3.3.3.3', 3, 'checking', '', NULL, 3, 1, ?, ?)`, ts, ts, ts, ts, ts, ts)
	chk := func(reg, cyc int, src, typ string, ok int, at string) {
		exec(`INSERT INTO checks (registry_id, cycle_id, ip_id, ip_address, attempt_number, validator_id, source, check_type, target, success, checked_at, created_at)
			VALUES (?, ?, ?, 'x', 1, '', ?, ?, 't', ?, ?, ?)`, reg, cyc, reg, src, typ, ok, at, at)
	}
	chk(1, 1, "egress", "https", 1, "2026-10-02T13:00:10Z")
	chk(1, 1, "inbound-site-1", "icmp", 1, "2026-10-02T13:00:20Z")
	chk(2, 1, "egress", "https", 1, "2026-10-02T13:05:00Z")
	chk(1, 2, "egress", "https", 0, "2026-10-02T16:00:10Z")
	exec(`INSERT INTO events (source_type, source_id, ip_id, event_type, payload, occurred_at, registry_id, cycle_id) VALUES
		('control-api', '', 1, 'fip_associated', '{"fip_id":"f","validator_id":"vkiplab-v5"}', ?, 1, 1),
		('control-api', '', 2, 'aggregated', '{"result":"pass","checks":1,"passed":1,"missing":1}', ?, 2, 1)`, ts, ts)
	raw.Close()

	d, err := Open(ctx, path)
	if err != nil {
		t.Fatalf("open (runs migration 11): %v", err)
	}
	defer d.Close()

	rs := runs(t, d)
	// run 1 and run 2 from the checks, plus the open run that adopted 3.3.3.3
	if len(rs) != 3 {
		t.Fatalf("expected 3 runs, got %+v", rs)
	}
	var first, second, open RunSummary
	for _, r := range rs {
		switch {
		case r.State == RunOpen:
			open = r
		case r.Addresses == 2:
			first = r
		default:
			second = r
		}
	}
	if first.Pass != 2 || first.Fail != 0 || second.Fail != 1 || second.Addresses != 1 || open.Total != 1 {
		t.Fatalf("runs: first=%+v second=%+v open=%+v", first, second, open)
	}
	res, _ := d.ListRunResults(ctx, first.ID)
	for _, r := range res {
		if r.IPAddress == "1.1.1.1" && (r.CycleID != 1 || r.Verdict != ResultPass || !r.Derived) {
			// the queue row holds the later cycle, so this one is derived from its checks
			t.Errorf("1.1.1.1 in the first run: %+v", r)
		}
		if r.IPAddress == "2.2.2.2" && (r.ExpectedChecks != 2 || r.RecordedChecks != 1 || r.Verdict != ResultPass || r.Derived) {
			t.Errorf("result from the aggregated event: %+v", r)
		}
	}
	res2, _ := d.ListRunResults(ctx, second.ID)
	if len(res2) != 1 || res2[0].CycleID != 2 || res2[0].Verdict != ResultFail || res2[0].Derived {
		t.Errorf("second run result: %+v", res2)
	}
	var v string
	if err := d.QueryRowContext(ctx, `SELECT validator_id FROM checks WHERE source='inbound-site-1'`).Scan(&v); err != nil || v != "vkiplab-v5" {
		t.Errorf("ingress validator = %q err=%v", v, err)
	}
	var unset int
	d.QueryRowContext(ctx, `SELECT COUNT(*) FROM checks WHERE run_id IS NULL`).Scan(&unset)
	if unset != 0 {
		t.Errorf("%d checks without a run", unset)
	}
	var ver int
	d.QueryRowContext(ctx, `PRAGMA user_version`).Scan(&ver)
	if ver != 12 {
		t.Errorf("user_version = %d", ver)
	}
}