package data

import (
	"database/sql"
	"fmt"
	"reflect"
)

// NewFromSQLRows returns a new Frame populated with the data from rows. The Field Vector types
// will be Vectors of pointer types, []*T, if the SQL column is nullable or if the nullable property is unknown.
// Otherwise, they will be []T types.
//
// Fields will be named to match name of the SQL columns and the SQL column names must be unique (https://github.com/grafana/grafana-plugin-sdk-go/issues/59).
//
// All the types must be supported by the Frame or a SQLStringConverter will be created and
// the resulting Field Vector type will be of type []*string.
//
// The SQLStringConverter's ConversionFunc will be applied to matching rows if it is not nil.
// Additionally, if the SQLStringConverter's Replacer is not nil, the replacement will be performed.
// A map of Field/Column index to the corresponding SQLStringConverter is returned so what conversions were
// done can be inspected.
func NewFromSQLRows(rows *sql.Rows, converters ...SQLStringConverter) (*Frame, map[int]SQLStringConverter, error) {
	frame, mappers, err := newForSQLRows(rows, converters...)
	if err != nil {
		return nil, nil, err
	}

	for rows.Next() {
		sRow := frame.newScannableRow()
		err := rows.Scan(sRow...)
		if err != nil {
			return nil, nil, err
		}
	}

	for fieldIdx, mapper := range mappers {
		if mapper.ConversionFunc == nil {
			continue
		}
		vec := frame.Fields[fieldIdx]
		for i := 0; i < vec.Len(); i++ {
			v, err := mapper.ConversionFunc(vec.vector.At(i).(*string))
			if err != nil {
				return nil, nil, err
			}
			vec.vector.Set(i, v)
		}
		if mapper.Replacer == nil {
			continue
		}
		if err := Replace(frame, fieldIdx, mapper.Replacer); err != nil {
			return nil, nil, err
		}
	}

	return frame, mappers, nil
}

// newForSQLRows creates a new Frame appropriate for scanning SQL rows with
// the the new Frame's ScannableRow() method.
func newForSQLRows(rows *sql.Rows, converters ...SQLStringConverter) (*Frame, map[int]SQLStringConverter, error) {
	mapping := make(map[int]SQLStringConverter)
	colTypes, err := rows.ColumnTypes()
	if err != nil {
		return nil, nil, err
	}
	colNames, err := rows.Columns()
	if err != nil {
		return nil, nil, err
	}
	// In the future we can probably remove this restriction. But right now we map names to Arrow Field Names.
	// Arrow Field names must be unique: https://github.com/grafana/grafana-plugin-sdk-go/issues/59
	seen := map[string]int{}
	for i, name := range colNames {
		if j, ok := seen[name]; ok {
			return nil, nil, fmt.Errorf(`duplicate column names are not allowed, found identical name "%v" at column indices %v and %v`, name, j, i)
		}
		seen[name] = i
	}
	frame := &Frame{}
	for i, colType := range colTypes {
		colName := colNames[i]
		nullable, ok := colType.Nullable()
		if !ok {
			nullable = true // If we don't know if it is nullable, assume it is
		}
		scanType := colType.ScanType()
		for _, converter := range converters {
			if converter.InputScanKind == scanType.Kind() && converter.InputTypeName == colType.DatabaseTypeName() {
				nullable = true // String converters are always nullable
				scanType = reflect.TypeOf("")
				mapping[i] = converter
			}
		}
		var vec interface{}
		if !nullable {
			vec = reflect.MakeSlice(reflect.SliceOf(scanType), 0, 0).Interface()
		} else {
			ptrType := reflect.TypeOf(reflect.New(scanType).Interface())
			// Nullabe types get passed to scan as a pointer to a pointer
			vec = reflect.MakeSlice(reflect.SliceOf(ptrType), 0, 0).Interface()
		}
		if !ValidFieldType(vec) {
			// Automatically create string mapper if we end up with an unsupported type
			mapping[i] = SQLStringConverter{
				Name:          fmt.Sprintf("Autogenerated for column %v", i),
				InputTypeName: colType.DatabaseTypeName(),
				InputScanKind: colType.ScanType().Kind(),
			}
			ptrType := reflect.TypeOf(reflect.New(reflect.TypeOf("")).Interface())
			vec = reflect.MakeSlice(reflect.SliceOf(ptrType), 0, 0).Interface()
		}
		frame.Fields = append(frame.Fields, NewField(colName, nil, vec))
	}
	return frame, mapping, nil
}

// newScannableRow adds a row to the Frame by extending each Field's Vector. It returns
// a slice of references that can be passed to the database/sql rows.Scan() to scan directly into
// the extended Vectors of the data.
func (f *Frame) newScannableRow() []interface{} {
	row := make([]interface{}, len(f.Fields))
	for i, field := range f.Fields {
		vec := field.vector
		vec.Extend(1)
		// non-nullable fields will be *T, and nullable fields will be **T
		vecItemPointer := vec.PointerAt(vec.Len() - 1)
		row[i] = vecItemPointer
	}
	return row
}

// SQLStringConverter can be used to store types not supported by
// a Frame into a *string. When scanning, if a SQL's row's InputScanType's Kind
// and InputScanKind match that returned by the sql response, then the
// conversion func will be run on the row.
type SQLStringConverter struct {
	// Name is an optional property that can be used to identify a converter
	Name          string
	InputScanKind reflect.Kind // reflect.Type might better or worse option?
	InputTypeName string

	// Conversion func may be nil to do no additional operations on the string conversion.
	ConversionFunc func(in *string) (*string, error)

	// If the Replacer is not nil, the replacement will be performed.
	Replacer *StringFieldReplacer
}

// Note: SQLStringConverter is perhaps better understood as []byte. However, currently
// the Vector type ([][]byte) is not supported. https://github.com/grafana/grafana-plugin-sdk-go/issues/57

// StringFieldReplacer is used to replace a *string Field in a data. The type
// returned by the ReplaceFunc must match the type of elements of VectorType.
// Both properties must be non-nil.
type StringFieldReplacer struct {
	VectorType  interface{}
	ReplaceFunc func(in *string) (interface{}, error)
}

// Replace will replace a *string Vector of the specified Field's index
// using the StringFieldReplacer.
func Replace(frame *Frame, fieldIdx int, replacer *StringFieldReplacer) error {
	if fieldIdx > len(frame.Fields) {
		return fmt.Errorf("fieldIdx is out of bounds, field len: %v", len(frame.Fields))
	}
	field := frame.Fields[fieldIdx]
	if field.Type() != FieldTypeNullableString {
		return fmt.Errorf("can only replace []*string vectors, vector is of type %s", field.Type())
	}

	if !ValidFieldType(replacer.VectorType) {
		return fmt.Errorf("can not replace column with unsupported type %T", replacer.VectorType)
	}
	newVector := newVector(replacer.VectorType, field.vector.Len())
	for i := 0; i < newVector.Len(); i++ {
		oldVal := field.vector.At(i).(*string) // Vector type is checked earlier above
		newVal, err := replacer.ReplaceFunc(oldVal)
		if err != nil {
			return err
		}
		newVector.Set(i, newVal)
	}
	field.vector = newVector
	return nil
}
