package main

import (
	"context"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"testing"
)

// The schema is numbered migrations and a consolidated schema for a fresh database (the mesh's rule: both). A
// database given the consolidated schema and one given every migration in order hold the same tables, columns,
// constraints and indexes.
func TestTheConsolidatedSchemaIsTheMigrationsApplied(t *testing.T) {
	ctx := context.Background()
	fresh := pgStore(t) // Prepare on an empty database: the consolidated schema
	stepped, err := OpenPgStore(ctx, freshDatabase(t))
	if err != nil {
		t.Fatal(err)
	}
	defer stepped.Close()
	migrations, err := Migrations()
	if err != nil {
		t.Fatal(err)
	}
	for _, m := range migrations {
		if _, err := stepped.pool.Exec(ctx, m.SQL); err != nil {
			t.Fatalf("migration %04d: %v", m.Number, err)
		}
	}
	a, b := shape(t, fresh), shape(t, stepped)
	if !reflect.DeepEqual(a, b) {
		t.Fatalf("the consolidated schema and the migrations differ:\nconsolidated %v\nmigrations   %v", a, b)
	}
	if len(a) < 20 {
		t.Fatalf("the schema read is too small to compare: %v", a)
	}
	// The consolidated schema says how far it is consolidated.
	if !strings.Contains(consolidated, "through migration "+itoa4(len(migrations))) {
		t.Fatalf("schema.sql does not say it is consolidated through migration %04d", len(migrations))
	}
}

func shape(t *testing.T, s *PgStore) []string {
	t.Helper()
	var out []string
	for _, q := range []string{
		`select table_name || '.' || column_name || ' ' || data_type || ' ' || is_nullable || ' ' || coalesce(column_default, '')
		   from information_schema.columns where table_schema = 'public' and table_name <> 'migration' order by 1`,
		`select conrelid::regclass::text || ' ' || pg_get_constraintdef(oid) from pg_constraint
		   where connamespace = 'public'::regnamespace and conrelid::regclass::text <> 'migration' order by 1`,
		`select indexdef from pg_indexes where schemaname = 'public' and tablename <> 'migration' order by 1`,
	} {
		rows, err := s.pool.Query(context.Background(), q)
		if err != nil {
			t.Fatal(err)
		}
		for rows.Next() {
			var line string
			if err := rows.Scan(&line); err != nil {
				t.Fatal(err)
			}
			out = append(out, line)
		}
		rows.Close()
	}
	return out
}

func TestPreparingTwiceChangesNothingAndAChangedMigrationIsRefused(t *testing.T) {
	ctx := context.Background()
	s := pgStore(t)
	said, err := s.Prepare(ctx)
	if err != nil || len(said) != 1 || !strings.Contains(said[0], "as this version needs") {
		t.Fatalf("a second preparation: %v %v", said, err)
	}
	if err := s.Ready(ctx); err != nil {
		t.Fatal(err)
	}
	if _, err := s.pool.Exec(ctx, `update migration set checksum = 'other' where number = 1`); err != nil {
		t.Fatal(err)
	}
	if _, err := s.Prepare(ctx); err == nil || !strings.Contains(err.Error(), "has changed since") {
		t.Fatalf("a changed migration: %v", err)
	}
}

func TestAStoreBehindItsMigrationsRefusesToServe(t *testing.T) {
	ctx := context.Background()
	s, err := OpenPgStore(ctx, freshDatabase(t))
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	if err := s.Ready(ctx); err == nil || !strings.Contains(err.Error(), "preparation step") {
		t.Fatalf("an unprepared database: %v", err)
	}
	l := &lazyStore{open: func(context.Context) (*PgStore, error) { return s, nil }}
	k := &KB{Store: l}
	if _, err := k.Get(GetAsk{Number: 276}); err == nil || !strings.Contains(err.Error(), "preparation step") {
		t.Fatalf("a verb over an unprepared database: %v", err)
	}
	if h := k.Health(l); h.Healthy || !strings.Contains(h.Why, "preparation step") {
		t.Fatalf("health over an unprepared database: %+v", h)
	}
}

// The migrations carried are numbered from 0001 without a gap, and every file in migrations/ is one.
func TestTheMigrationsAreNumberedWithoutAGap(t *testing.T) {
	ms, err := Migrations()
	if err != nil {
		t.Fatal(err)
	}
	files, _ := os.ReadDir("migrations")
	if len(ms) == 0 || len(ms) != len(files) {
		t.Fatalf("%d migrations, %d files", len(ms), len(files))
	}
	for _, f := range files {
		if filepath.Ext(f.Name()) != ".sql" {
			t.Fatalf("%s is not a migration", f.Name())
		}
	}
}

// The daemons' lines the rules refuse are the catalogue's own copy.
func TestTheDaemonLinesAreTheCataloguesCopy(t *testing.T) {
	theirs := DaemonLines(readFile(t, filepath.Join("..", "..", "..", "..", "retired-lines")))
	ours := DaemonLines(daemonLinesFile)
	if len(ours) < 50 || !reflect.DeepEqual(theirs, ours) {
		t.Fatalf("daemon-lines differs from the catalogue's retired-lines (%d against %d)", len(ours), len(theirs))
	}
}