package bridge

import (
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"os"
	"sort"

	nodev1 "k8s.io/api/node/v1"

	_ "github.com/mattn/go-sqlite3"
)

// ErrStoreNotFound is returned when the local SQLite store file does not
// exist.
var ErrStoreNotFound = errors.New("local store not found")

// Store reads Kubernetes API objects from the KubeEdge local SQLite store
// (/var/lib/kubeedge/edgecore.db by default).
type Store struct {
	path string
	db   *sql.DB
}

// OpenStore opens the KubeEdge local SQLite store read-only.
func OpenStore(path string) (*Store, error) {
	if _, err := os.Stat(path); err != nil {
		if os.IsNotExist(err) {
			return nil, fmt.Errorf("%w: %s", ErrStoreNotFound, path)
		}
		return nil, err
	}
	db, err := sql.Open("sqlite3", "file:"+path+"?mode=ro")
	if err != nil {
		return nil, err
	}
	if err := db.Ping(); err != nil {
		_ = db.Close()
		return nil, err
	}
	return &Store{path: path, db: db}, nil
}

// Close closes the underlying database handle.
func (s *Store) Close() error {
	return s.db.Close()
}

// ListRuntimeClasses returns the RuntimeClass objects stored locally.
// Both the meta_v2 table (key format /<group>/<version>/<resource>/<ns>/<name>)
// and the legacy meta table (key format <ns>/<type>/<name>) are scanned, so
// the tool keeps working regardless of which store generation the node runs.
// Column naming varies between KubeEdge versions (group_version_resource vs
// groupversionresource), so the schema is introspected first.
func (s *Store) ListRuntimeClasses() ([]nodev1.RuntimeClass, error) {
	classes := []nodev1.RuntimeClass{}
	seen := map[string]bool{}
	classes, seen = s.scanTable("meta_v2", classes, seen)
	classes, _ = s.scanTable("meta", classes, seen)
	sort.Slice(classes, func(i, j int) bool { return classes[i].Name < classes[j].Name })
	return classes, nil
}

func (s *Store) scanTable(table string, classes []nodev1.RuntimeClass, seen map[string]bool) ([]nodev1.RuntimeClass, map[string]bool) {
	columns, err := s.tableColumns(table)
	if err != nil || !hasColumn(columns, "value") {
		return classes, seen
	}
	where := "key LIKE '%/runtimeclass%'"
	if hasColumn(columns, "group_version_resource") {
		where += " OR group_version_resource LIKE '%runtimeclass%'"
	} else if hasColumn(columns, "groupversionresource") {
		where += " OR groupversionresource LIKE '%runtimeclass%'"
	} else if hasColumn(columns, "type") {
		where += " OR type LIKE '%runtimeclass%'"
	}
	rows, err := s.db.Query(fmt.Sprintf("SELECT value FROM %s WHERE %s", table, where))
	if err != nil {
		return classes, seen
	}
	return appendClasses(rows, classes, seen)
}

func (s *Store) tableColumns(table string) ([]string, error) {
	rows, err := s.db.Query(fmt.Sprintf("SELECT name FROM pragma_table_info('%s')", table))
	if err != nil {
		return nil, err
	}
	defer func() { _ = rows.Close() }()
	var columns []string
	for rows.Next() {
		var name string
		if err := rows.Scan(&name); err != nil {
			return nil, err
		}
		columns = append(columns, name)
	}
	return columns, rows.Err()
}

func hasColumn(columns []string, name string) bool {
	for _, c := range columns {
		if c == name {
			return true
		}
	}
	return false
}

func appendClasses(rows *sql.Rows, classes []nodev1.RuntimeClass, seen map[string]bool) ([]nodev1.RuntimeClass, map[string]bool) {
	defer func() { _ = rows.Close() }()
	for rows.Next() {
		var raw string
		if err := rows.Scan(&raw); err != nil {
			continue
		}
		var probe struct {
			Kind string `json:"kind"`
		}
		if err := json.Unmarshal([]byte(raw), &probe); err != nil {
			continue
		}
		if probe.Kind != "RuntimeClass" {
			continue
		}
		var rc nodev1.RuntimeClass
		if err := json.Unmarshal([]byte(raw), &rc); err != nil {
			continue
		}
		if rc.Name == "" || seen[rc.Name] {
			continue
		}
		seen[rc.Name] = true
		classes = append(classes, rc)
	}
	return classes, seen
}