dbmeta/save.go

package dbmeta

import (
	"context"
	"fmt"
	"reflect"
	"strings"

	"git.bit5.ru/backend/db"
	"git.bit5.ru/backend/errors"
	"git.bit5.ru/backend/meta"

	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	"go.opentelemetry.io/otel/trace"
)

const tracerName = "git.bit5.ru/backend/dbmeta"

var tracer = otel.Tracer(tracerName)

type SaveIncompatibleError struct {
	s string
}

func (e *SaveIncompatibleError) Error() string {
	return e.s
}

type dataRowInfo struct {
	mType     reflect.Type
	data      meta.IMetaStruct
	pkey      string
	tableName string
	fields    []string
	mapFields map[int]int
}

type RemovedIds struct {
	//class_id => []ids
	ids map[uint32][]uint64
}

func SaveRow(ctx context.Context, db *db.DBC, dataItem meta.IMetaStruct) error {

	ctx, span := tracer.Start(ctx, "SaveRow")
	defer span.End()

	info, err := makeDataRowInfo(dataItem)
	if err != nil {
		return err
	}

	fields := make([]string, 0, len(info.fields))
	params := make([]interface{}, 0, len(info.fields))
	elem := reflect.ValueOf(dataItem).Elem()

	var pkey *reflect.Value

	bitmaskItem, bitmaskable := dataItem.(meta.IBitmasked)

	for fieldIdx, fieldNum := range info.mapFields {
		value := elem.Field(fieldNum)
		if info.fields[fieldIdx] == info.pkey && isEmptyField(value) {
			pkey = &value
			continue
		}

		skipAdding := bitmaskable && bitmaskItem.IsMaskFilled() && !bitmaskItem.HasValue(uint64(fieldIdx))
		if skipAdding {
			continue
		}

		fields = append(fields, info.fields[fieldIdx])
		params = append(params, value.Interface())
	}

	if len(fields) == 0 {
		return errors.New("Fields list is empty")
	}

	tableName := info.tableName
	sqlSmt := createInsertSQLForFields(ctx, tableName, fields, 1 /*one_row*/)

	span.SetAttributes(
		attribute.String("table", tableName),
		attribute.String("sql", sqlSmt),
		//semconv.DBSQLTableKey.String(tableName),
		//semconv.DBStatementKey.String(sqlSmt),
	)

	updateBuilder := db.UpdateBySQL(sqlSmt, params...)

	res, err := updateBuilder.ExecContext(ctx)
	if err != nil {
		if len(sqlSmt) > 200 {
			sqlSmt = sqlSmt[0:200]
		}
		return errors.Errorf("Can not execute SaveRow. Got error from updateBuilder.ExecContext. %s (%s)", err.Error(), sqlSmt)
	}

	insertId, _ := res.LastInsertId()
	if pkey != nil && insertId != 0 {
		pkey.Set(reflect.ValueOf(uint32(insertId)).Convert(pkey.Type()))
	}

	return nil
}

func SaveMetaRootStruct(
	ctx context.Context,
	db *db.DBC,
	data meta.IMetaStruct,
	ownerId uint32,
	removedIds meta.IRemovedIds,
	deltaSave bool,
) error {

	spanAttrs := trace.WithAttributes(
		attribute.Int64("owner.id", int64(ownerId)),
	)

	ctx, span := tracer.Start(ctx, "SaveMetaRootStruct", spanAttrs)
	defer span.End()

	dataItem := reflect.ValueOf(data)
	mType := dataItem.Type()
	if mType.Kind() == reflect.Ptr {
		dataItem = dataItem.Elem()
		mType = mType.Elem()
	}

	var fprops map[string]map[string]string
	if dataItem.Addr().CanInterface() {
		imeta, _ := dataItem.Addr().Interface().(meta.IClassProps)
		if imeta != nil {
			fprops = *imeta.CLASS_FIELDS_PROPS()
		}
	}

	bitmaskItem, bitmaskable := data.(meta.IBitmasked)

	numField := mType.NumField()

	span.SetAttributes(
		attribute.String("mType.name", mType.Name()),
		attribute.Int("mType.numField", numField),
	)

	for i := 0; i < numField; i++ {
		skipAdding := bitmaskable && bitmaskItem.IsMaskFilled() && !bitmaskItem.HasValue(uint64(i))
		if skipAdding {
			continue
		}

		field := mType.Field(i)
		tfield := field.Type
		// Checking if this field should be skipped
		if fprops != nil {
			if props, ok := fprops[field.Name]; ok {
				if _, ok := props["db_skip_save"]; ok {
					continue
				}
			}
		}
		// skip unexported fields
		if len(field.PkgPath) > 0 {
			continue
		}
		switch tfield.Kind() {
		case reflect.Slice:
			if err := SaveMetaCollection(ctx, db, dataItem.Field(i), ownerId, removedIds, deltaSave); err != nil {
				span.RecordError(err)
				span.SetStatus(codes.Error, err.Error())
				return errors.WithMessagef(err, "Can not execute SaveMetaRootStruct. Got error from SaveMetaCollection. ownerId: %d.", ownerId)
			}
			break
		case reflect.Struct:
			if err := saveStruct(ctx, db, dataItem.Field(i), ownerId); err != nil {
				span.RecordError(err)
				span.SetStatus(codes.Error, err.Error())
				return errors.WithMessagef(err, "Can not execute SaveMetaRootStruct. Got error from saveStruct. ownerId: %d.", ownerId)
			}
			break
		case reflect.Map:
		case reflect.Ptr:
		case reflect.Array:
		case reflect.UnsafePointer:
			return errors.Errorf("Unsupported type: %s", tfield.Kind())
		default:
			break
		}
	}

	return nil
}

func saveStruct(
	ctx context.Context,
	db *db.DBC,
	dataItem reflect.Value,
	ownerId uint32,
) error {

	ctx, span := tracer.Start(ctx, "saveStruct")
	defer span.End()

	mType := dataItem.Type()
	if mType.Kind() == reflect.Ptr {
		dataItem = dataItem.Elem()
		mType = mType.Elem()
	}

	var row meta.IMetaStruct
	if dataItem.Addr().CanInterface() {
		row, _ = dataItem.Addr().Interface().(meta.IMetaStruct)
	}
	if row == nil {
		return errors.Errorf("Couldn't convert to IMetaStruct: %s", mType.Name())
	}

	props := *row.CLASS_PROPS()
	if ownerField, ok := props["owner"]; ok {
		field := dataItem.FieldByName(strings.Title(ownerField))
		if !field.IsValid() {
			return errors.Errorf("Owner field \"%s\" is not found in struct \"%s\"", ownerField, mType.Name())
		}

		// enforcing ownerId
		if field.CanSet() {
			field.SetUint(uint64(ownerId))
		}

	}

	if err := SaveRow(ctx, db, row); err != nil {
		span.RecordError(err)
		span.SetStatus(codes.Error, err.Error())
		return errors.WithMessagef(err, "Can not execute saveStruct. Got error from SaveRow. ownerId: %d.", ownerId)
	}

	return nil
}

func SaveMetaCollection(
	ctx context.Context,
	db *db.DBC,
	slice reflect.Value,
	ownerId uint32,
	removedIds meta.IRemovedIds,
	deltaSave bool,
) error {

	ctx, span := tracer.Start(ctx, "SaveMetaCollection")
	defer span.End()

	if slice.Type().Kind() != reflect.Slice {
		err := errors.Errorf("It isn't slice: %s", slice.Type().Kind())
		span.RecordError(err)
		span.SetStatus(codes.Error, err.Error())
		return err
	}

	sliceItem := reflect.New(slice.Type().Elem().Elem())
	if sliceItem.Type().Kind() != reflect.Ptr {
		sliceItem = sliceItem.Addr()
	}

	var row meta.IMetaStruct
	if sliceItem.CanInterface() {
		row, _ = sliceItem.Interface().(meta.IMetaStruct)
	}

	if row == nil {
		err := errors.Errorf("Couldn't convert to IMetaStruct: %s", sliceItem.Kind())
		span.RecordError(err)
		span.SetStatus(codes.Error, err.Error())
		return err
	}

	item, _ := sliceItem.Interface().(meta.IMetaDataItem)
	if item == nil {
		err := errors.Errorf("Couldn't convert to IMetaDataItem: %s", slice.Type())
		span.RecordError(err)
		span.SetStatus(codes.Error, err.Error())
		return err
	}

	if !deltaSave {
		cond := fmt.Sprintf("`%s`=%d ", item.GetOwnerFieldName(), ownerId)
		_, err := db.DeleteFrom(item.GetDbTableName()).Where(cond).Exec()
		if err != nil {
			span.RecordError(err)
			span.SetStatus(codes.Error, err.Error())
			return err
		}
	}

	collection, err := NewDataCollection(db, ownerId, "", item)
	if err != nil {
		span.RecordError(err)
		span.SetStatus(codes.Error, err.Error())
		return err
	}

	for ind := 0; ind < slice.Len(); ind++ {
		ind_item, err := convertToIdataItem(slice.Index(ind))
		if err != nil {
			span.RecordError(err)
			span.SetStatus(codes.Error, err.Error())
			return err
		}
		// NOTE: we don't check here for error on purpose, here it's considered to be OK
		collection.Checkin(ind_item)
	}

	if err := collection.Save(ctx); err != nil {
		span.RecordError(err)
		span.SetStatus(codes.Error, err.Error())
		return err
	}

	if removedIds != nil {
		classId := item.CLASS_ID()

		if removedIds.HasList(classId) {
			err := deleteByIds(db, ownerId, item, removedIds.GetList(classId))
			if err != nil {
				span.RecordError(err)
				span.SetStatus(codes.Error, err.Error())
				return err
			}
		}
	}
	return nil
}

func deleteByIds(db *db.DBC, ownerId uint32, item meta.IMetaDataItem, removedIds []uint64) error {
	var cond string

	if item.GetOwnerFieldName() == item.GetIdFieldName() {
		cond = fmt.Sprintf("`%s`=%d ", item.GetOwnerFieldName(), ownerId)
	} else {
		if len(removedIds) == 0 {
			return nil
		}
		ids := make([]string, 0, len(removedIds))
		for _, id := range removedIds {
			ids = append(ids, fmt.Sprintf("%d", id))
		}
		cond = fmt.Sprintf(
			"`%s`=%d AND `%s` IN(%s)",
			item.GetOwnerFieldName(),
			ownerId,
			item.GetIdFieldName(),
			strings.Join(ids, ","))
	}

	_, err := db.DeleteFrom(item.GetDbTableName()).Where(cond).Exec()
	return err
}

func convertToIdataItem(elem reflect.Value) (meta.IMetaDataItem, error) {
	mType := elem.Type()
	if mType.Kind() != reflect.Ptr {
		elem = elem.Addr()
		mType = elem.Type()
	}
	item, _ := elem.Interface().(meta.IMetaDataItem)
	if item == nil {
		return nil, errors.Errorf("Couldn't convert to IMetaDataItem: %s", mType.Name())
	}
	return item, nil
}

func convertToIdataRow(elem reflect.Value) (meta.IMetaStruct, error) {
	mType := elem.Type()
	if mType.Kind() != reflect.Ptr {
		elem = elem.Addr()
		mType = elem.Type()
	}
	item, _ := elem.Interface().(meta.IMetaStruct)
	if item == nil {
		return nil, errors.Errorf("Couldn't convert to IdataRow: %s", mType.Name())
	}
	return item, nil
}

func makeDataRowInfo(data meta.IMetaStruct) (*dataRowInfo, error) {
	info := new(dataRowInfo)
	info.mType = reflect.TypeOf(data).Elem()
	props := *data.CLASS_PROPS()
	tableName, ok := props["table"]
	info.tableName = tableName
	if !ok || len(info.tableName) == 0 {
		return nil, errors.Errorf("Prop 'table' is empty (%s)", info.mType.Name())
	}
	pkey, ok := props["pkey"]
	info.pkey = ""
	if ok {
		info.pkey = pkey
	}
	info.fields = data.CLASS_FIELDS()
	info.mapFields = make(map[int]int, len(info.fields))
	for ind, field := range info.fields {
		found := false
		for i := 0; i < info.mType.NumField(); i++ {
			if strings.Title(field) == info.mType.Field(i).Name {
				found = true
				info.mapFields[ind] = i
			}
		}
		if !found {
			return nil, errors.Errorf("Field %s[%d] not found in struct %s", field, ind, info.mType.Name())
		}
	}
	return info, nil
}

func isEmptyField(v reflect.Value) bool {
	vint := reflect.ValueOf(int64(0))
	vstr := reflect.ValueOf("")
	if v.Type().ConvertibleTo(vint.Type()) {
		return reflect.DeepEqual(v.Convert(vint.Type()).Interface(), vint.Interface())
	}
	return reflect.DeepEqual(v.Interface(), vstr.Interface())
}

func (rids *RemovedIds) init() {
	if rids.ids == nil {
		rids.ids = make(map[uint32][]uint64)
	}
}

func (rids *RemovedIds) GetList(classId uint32) []uint64 {
	rids.init()
	if list, ok := rids.ids[classId]; ok {
		return list
	}
	return []uint64{}
}

func (rids *RemovedIds) Add(classId uint32, id uint64) {
	rids.init()
	if _, ok := rids.ids[classId]; !ok {
		rids.ids[classId] = []uint64{}
	}
	rids.ids[classId] = append(rids.ids[classId], id)
}

func (rids *RemovedIds) HasList(classId uint32) bool {
	rids.init()
	_, ok := rids.ids[classId]
	return ok
}

func GetChangedRootStructFields(data meta.IMetaStruct, fieldNames []string) map[string]reflect.Value {
	var fields = make(map[string]reflect.Value)

	dataItem := reflect.ValueOf(data)
	mType := dataItem.Type()
	if mType.Kind() == reflect.Ptr {
		dataItem = dataItem.Elem()
		mType = mType.Elem()
	}

	bitmaskItem, bitmaskable := data.(meta.IBitmasked)

	for i := 0; i < mType.NumField(); i++ {
		fieldName := mType.Field(i).Name

		if !sliceOfStringsContains(fieldNames, fieldName) {
			continue
		}

		skipAdding := bitmaskable && bitmaskItem.IsMaskFilled() && !bitmaskItem.HasValue(uint64(i))
		if skipAdding {
			continue
		}

		fields[fieldName] = dataItem.Field(i)
	}

	return fields
}

func sliceOfStringsContains(slice []string, v string) bool {
	for _, item := range slice {
		if item == v {
			return true
		}
	}
	return false
}

func GetChangedStructFields(dataItem reflect.Value, fieldNamesFilter func(string) bool) (map[string]reflect.Value, error) {
	var fields = make(map[string]reflect.Value)

	mType := dataItem.Type()
	if mType.Kind() == reflect.Ptr {
		dataItem = dataItem.Elem()
		mType = mType.Elem()
	}

	var row meta.IMetaStruct
	if dataItem.Addr().CanInterface() {
		row, _ = dataItem.Addr().Interface().(meta.IMetaStruct)
	}
	if row == nil {
		return nil, errors.Errorf("Couldn't convert to IMetaStruct: %s", mType.Name())
	}

	info, err := makeDataRowInfo(row)
	if err != nil {
		return nil, err
	}

	elem := reflect.ValueOf(row).Elem()
	bitmaskItem, bitmaskable := row.(meta.IBitmasked)

	for fieldIdx, fieldNum := range info.mapFields {
		fieldName := info.fields[fieldIdx]
		if fieldNamesFilter != nil && !fieldNamesFilter(fieldName) {
			continue
		}

		value := elem.Field(fieldNum)

		if fieldName == info.pkey && isEmptyField(value) {
			continue
		}

		skipAdding := bitmaskable && bitmaskItem.IsMaskFilled() && !bitmaskItem.HasValue(uint64(fieldIdx))
		if skipAdding {
			continue
		}

		fields[fieldName] = value
	}

	return fields, nil
}