dps-sql/v1.go

package dpsapi

import (
	"errors"
	"fmt"
	"git.sxidc.com/go-tools/api_binding/http_binding"
	"git.sxidc.com/go-tools/api_binding/http_binding/binding_context"
	"git.sxidc.com/go-tools/api_binding/http_binding/response"
	"git.sxidc.com/go-tools/api_binding/utils"
	"git.sxidc.com/service-supports/dps-sdk"
	"git.sxidc.com/service-supports/dps-sdk/client"
	"time"
)

func ApiV1(binding *http_binding.Binding, dpsAddress string, operatorIDFunc OperatorIDFunc) {
	if binding == nil {
		panic("没有传递http_binding")
	}

	if utils.IsStringEmpty(dpsAddress) {
		panic("没有指定dps地址")
	}

	if operatorIDFunc == nil {
		panic("没有传递获取operatorID的回调函数")
	}

	http_binding.PostBind(binding, &http_binding.SimpleBindItem[OperateParseRequest, map[string]any]{
		Path:         "/dpsv1/database/operate/parse",
		ResponseFunc: response.SendMapResponse,
		BusinessFunc: func(c *binding_context.Context, inputModel OperateParseRequest) (map[string]any, error) {
			parsedClauses, err := parseSql(inputModel.SQL)
			if err != nil {
				return nil, err
			}

			for _, parsedClause := range parsedClauses {
				switch clause := parsedClause.(type) {
				case *insertClause:
					return map[string]any{
						"parsed": insertMap(clause),
					}, nil
				case *deleteClause:
					return map[string]any{
						"parsed": deleteMap(clause),
					}, nil
				case *updateClause:
					return map[string]any{
						"parsed": updateMap(clause),
					}, nil
				case *selectClause:
					return map[string]any{
						"parsed": selectMap(clause),
					}, nil
				default:
					return nil, errors.New("不支持的SQL语句")
				}
			}

			return nil, nil
		},
	})

	http_binding.PostBind(binding, &http_binding.SimpleBindItem[OperateRequest, map[string]any]{
		Path:         "/dpsv1/database/operate",
		ResponseFunc: response.SendMapResponse,
		BusinessFunc: func(c *binding_context.Context, inputModel OperateRequest) (map[string]any, error) {
			version := "v1"
			if utils.IsStringNotEmpty(inputModel.Version) {
				version = inputModel.Version
			}

			keyColumns := []string{"id"}
			if inputModel.KeyColumns != nil && len(inputModel.KeyColumns) != 0 {
				keyColumns = inputModel.KeyColumns
			}

			operatorID, err := operatorIDFunc(c)
			if err != nil {
				return map[string]any{"table_rows": make([]map[string]any, 0)}, err
			}

			parsedClauses, err := parseSql(inputModel.SQL)
			if err != nil {
				return map[string]any{"table_rows": make([]map[string]any, 0)}, err
			}

			dpsClient, err := dps.NewClient(dpsAddress, "v1", inputModel.DatabaseID)
			if err != nil {
				return map[string]any{"table_rows": make([]map[string]any, 0)}, err
			}

			result := make([]map[string]any, 0)

			err = dpsClient.Transaction(func(tx client.Transaction) error {
				for _, parsedClause := range parsedClauses {
					switch clause := parsedClause.(type) {
					case *insertClause:
						return doInsert(tx, version, keyColumns, clause, operatorID)
					case *deleteClause:
						return doDelete(tx, version, keyColumns, clause, operatorID)
					case *updateClause:
						return doUpdate(tx, version, keyColumns, clause, operatorID)
					case *selectClause:
						r, err := doSelect(dpsClient, version, clause)
						if err != nil {
							return err
						}

						result = r
						return nil
					default:
						return errors.New("不支持的SQL语句")
					}
				}

				return nil
			})
			if err != nil {
				return map[string]any{"table_rows": make([]map[string]any, 0)}, err
			}

			return map[string]any{"table_rows": result}, nil
		},
	})
}

func insertMap(clause *insertClause) map[string]any {
	tableRows := make(map[string]any)
	for columnName, value := range clause.tableRow {
		tableRows[columnName] = value.value
	}

	return map[string]any{
		"table":     clause.table,
		"table_row": tableRows,
	}
}

func deleteMap(clause *deleteClause) map[string]any {
	return map[string]any{
		"table": clause.table,
		"where": clause.where,
	}
}

func updateMap(clause *updateClause) map[string]any {
	newTableRows := make(map[string]any)
	for columnName, value := range clause.newTableRow {
		newTableRows[columnName] = value.value
	}

	return map[string]any{
		"table":         clause.table,
		"where":         clause.where,
		"new_table_row": newTableRows,
	}
}

func selectMap(clause *selectClause) map[string]any {
	return map[string]any{}
}

func doInsert(tx client.Transaction, version string, keyColumns []string, clause *insertClause, operatorID string) error {
	tableRow, err := clauseTableRowsToDPSTableRow(clause.tableRow)
	if err != nil {
		return err
	}

	statement, err := tx.InsertTx(&client.InsertRequest{
		TablePrefixWithSchema: clause.table,
		Version:               version,
		KeyColumns:            keyColumns,
		TableRow:              tableRow,
		UserID:                operatorID,
	})
	if err != nil {
		fmt.Println(statement)
		return err
	}

	return nil
}

func doDelete(tx client.Transaction, version string, keyColumns []string, clause *deleteClause, operatorID string) error {
	statement, err := tx.DeleteWhereTx(&client.DeleteWhereRequest{
		TablePrefixWithSchema: clause.table,
		Version:               version,
		KeyColumns:            keyColumns,
		Where:                 client.NewClause().Common(clause.where),
		UserID:                operatorID,
	})
	if err != nil {
		fmt.Println(statement)
		return err
	}

	return nil
}

func doUpdate(tx client.Transaction, version string, keyColumns []string, clause *updateClause, operatorID string) error {
	newTableRow, err := clauseTableRowsToDPSTableRow(clause.newTableRow)
	if err != nil {
		return err
	}

	statement, err := tx.UpdateWhereTx(&client.UpdateWhereRequest{
		TablePrefixWithSchema: clause.table,
		Version:               version,
		KeyColumns:            keyColumns,
		Where:                 client.NewClause().Common(clause.where),
		NewTableRow:           newTableRow,
		UserID:                operatorID,
	})
	if err != nil {
		fmt.Println(statement)
		return err
	}

	return nil
}

func doSelect(dpsClient client.Client, version string, clause *selectClause) ([]map[string]any, error) {
	statement, tableRows, err := dpsClient.CommonQueryOnly(&client.CommonQueryRequest{
		TablePrefixWithSchema: "",
		Table:                 "",
		Version:               version,
		Select:                nil,
		Where:                 nil,
		OrderBy:               nil,
		GroupBy:               nil,
		Joins:                 nil,
		Having:                nil,
		PageNo:                0,
		PageSize:              0,
	})
	if err != nil {
		fmt.Println(statement)
		return nil, err
	}

	results := make([]map[string]any, 0)
	for _, tableRow := range tableRows {
		results = append(results, tableRow.ToMap())
	}

	return results, nil
}

func clauseTableRowsToDPSTableRow(clauseTableRow map[string]clauseTableRowValue) (*client.TableRow, error) {
	tableRow := client.NewTableRow()

	for columnName, value := range clauseTableRow {
		switch value.kind {
		case clauseTableRowValueKindTime:
			tableRow.AddColumnValueTime(columnName, value.value.(time.Time))
		case clauseTableRowValueKindBool:
			tableRow.AddColumnValueBool(columnName, value.value.(bool))
		case clauseTableRowValueKindString:
			tableRow.AddColumnValueString(columnName, value.value.(string))
		case clauseTableRowValueKindUint64:
			tableRow.AddColumnValueUint64(columnName, value.value.(uint64))
		case clauseTableRowValueKindFloat64:
			tableRow.AddColumnValueFloat64(columnName, value.value.(float64))
		default:
			return nil, errors.New("不支持的值类型")
		}
	}

	return tableRow, nil
}