baize/infrastructure/database/data_service/transaction.go

package data_service

import (
	"encoding/json"
	"git.sxidc.com/go-framework/baize/infrastructure/database/data_service/grpc_client/v1"
	"git.sxidc.com/go-framework/baize/infrastructure/database/data_service/grpc_client/v1/request"
	"git.sxidc.com/go-framework/baize/infrastructure/database/sql"
	"git.sxidc.com/service-supports/fserr"
)

type TxFunc func(tx *Transaction) error

type Transaction struct {
	stream v1.SqlService_TransactionClient
}

func (tx *Transaction) ExecuteRawSql(sqlStr string, executeParams map[string]any) ([]sql.Result, error) {
	var retErr error

	defer func() {
		if retErr != nil {
			innerErr := tx.stream.CloseSend()
			if innerErr != nil {
				panic(innerErr)
			}
		}
	}()

	executeParamsJsonBytes, err := json.Marshal(executeParams)
	if err != nil {
		retErr = err
		return nil, retErr
	}

	err = tx.stream.SendMsg(&request.TransactionOperation{
		Request: &request.TransactionOperation_ExecuteRawSqlRequest{
			ExecuteRawSqlRequest: &request.ExecuteRawSqlRequest{
				SQL:           sqlStr,
				ExecuteParams: string(executeParamsJsonBytes),
			},
		},
	})
	if err != nil {
		retErr = err
		return nil, retErr
	}

	resp, err := tx.stream.Recv()
	if err != nil {
		retErr = err
		return nil, retErr
	}

	if !resp.Success {
		retErr = fserr.New(resp.Msg)
		return nil, retErr
	}

	tableRows := make([]map[string]any, 0)
	err = json.Unmarshal([]byte(resp.Results), &tableRows)
	if err != nil {
		retErr = err
		return nil, retErr
	}

	results := make([]sql.Result, len(tableRows))
	for i, row := range tableRows {
		results[i] = row
	}

	return results, nil
}

func (tx *Transaction) ExecuteSql(name string, executeParams map[string]any) ([]sql.Result, error) {
	var retErr error

	defer func() {
		if retErr != nil {
			innerErr := tx.stream.CloseSend()
			if innerErr != nil {
				panic(innerErr)
			}
		}
	}()

	executeParamsJsonBytes, err := json.Marshal(executeParams)
	if err != nil {
		retErr = err
		return nil, retErr
	}

	err = tx.stream.Send(&request.TransactionOperation{
		Request: &request.TransactionOperation_ExecuteSqlRequest{
			ExecuteSqlRequest: &request.ExecuteSqlRequest{
				Name:          name,
				ExecuteParams: string(executeParamsJsonBytes),
			},
		},
	})
	if err != nil {
		retErr = err
		return nil, retErr
	}

	resp, err := tx.stream.Recv()
	if err != nil {
		retErr = err
		return nil, retErr
	}

	if !resp.Success {
		retErr = fserr.New(resp.Msg)
		return nil, retErr
	}

	tableRows := make([]map[string]any, 0)
	err = json.Unmarshal([]byte(resp.Results), &tableRows)
	if err != nil {
		retErr = err
		return nil, retErr
	}

	results := make([]sql.Result, len(tableRows))
	for i, row := range tableRows {
		results[i] = row
	}

	return results, nil
}