<commit_before>package orm

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

	c "github.com/d0ngw/go/common"
)

// OpTxFunc 在事务中处理的函数
type OpTxFunc func(tx *sql.Tx) (interface{}, error)

// OpCreator Op
type OpCreator interface {
	//NewOp create a new Op
	NewOp() (*Op, error)
}

// Op 数据库操作接口,与sql.DB对应,封装了事务等
type Op struct {
	pool           *Pool          //数据连接
	tx             *sql.Tx        //事务
	txDone         bool           //事务是否结束
	rollbackOnly   bool           //是否只回滚
	transDepth     int            //调用的深度
	sharDBSerevcie ShardDBService //分片服务
}

// DB sql.DB
func (p *Op) DB() *sql.DB {
	return p.pool.db
}

// Pool pool
func (p *Op) Pool() *Pool {
	return p.pool
}

// PoolName name of pool
func (p *Op) PoolName() string {
	return p.pool.name
}

// SetupTableShard use op pool setup entity table shard
func (p *Op) SetupTableShard(entity Entity, ruleName string) error {
	if p.sharDBSerevcie == nil {
		return errors.New("no shard db service")
	}
	poolName, err := p.sharDBSerevcie.setupTableShard(entity, ruleName)
	if err != nil {
		return err
	}
	if poolName != p.PoolName() {
		return fmt.Errorf("op.PoolName %s != entity.PoolName %s", p.PoolName(), poolName)
	}
	return nil
}

func (p *Op) close() {
	p.tx = nil
	p.rollbackOnly = false
	p.transDepth = 0
}

//检查事务的状态
func (p *Op) checkTransStatus() error {
	if p.txDone {
		return sql.ErrTxDone
	}
	if p.tx == nil {
		return NewDBError(nil, "Not begin transaction")
	}
	return nil
}

func (p *Op) incrTransDepth() {
	p.transDepth = p.transDepth + 1
	c.Debugf("p.tranDepth:%v", p.transDepth)
}

func (p *Op) decrTransDepth() error {
	p.transDepth = p.transDepth - 1
	c.Debugf("p.tranDepth:%v", p.transDepth)
	if p.transDepth < 0 {
		return NewDBError(nil, "Too many invoke commit or rollback")
	}
	return nil
}

//结束事务
func (p *Op) finishTrans() error {
	if err := p.checkTransStatus(); err != nil {
		return err
	}
	if err := p.decrTransDepth(); err != nil {
		return err
	}
	if p.transDepth > 0 {
		return nil
	}
	defer p.close()
	p.txDone = true
	if p.rollbackOnly {
		c.Debugf("Rollback")
		return p.tx.Rollback()
	}
	c.Debugf("Commit")
	return p.tx.Commit()
}

// BeginTx 开始事务,支持简单的嵌套调用,如果已经开始了事务,则直接返回成功
func (p *Op) BeginTx() (err error) {
	p.incrTransDepth()
	if p.tx != nil {
		return nil //事务已经开启
	}
	if tx, err := p.DB().Begin(); err == nil {
		p.tx = tx
		return nil
	}
	return err
}

// Commit 提交事务
func (p *Op) Commit() error {
	return p.finishTrans()
}

// Rollback 回滚事务
func (p *Op) Rollback() error {
	p.SetRollbackOnly(true)
	return p.finishTrans()
}

// SetRollbackOnly 设置只回滚
func (p *Op) SetRollbackOnly(rollback bool) {
	p.rollbackOnly = rollback
}

// IsRollbackOnly 是否只回滚
func (p *Op) IsRollbackOnly() bool {
	return p.rollbackOnly
}

// DoInTrans 在事务中执行<commit_msg>fix transaction<commit_after>package orm

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

	c "github.com/d0ngw/go/common"
)

// OpTxFunc 在事务中处理的函数
type OpTxFunc func(tx *sql.Tx) (interface{}, error)

// OpCreator Op
type OpCreator interface {
	//NewOp create a new Op
	NewOp() (*Op, error)
}

// Op 数据库操作接口,与sql.DB对应,封装了事务等
type Op struct {
	pool           *Pool          //数据连接
	tx             *sql.Tx        //事务
	txDone         bool           //事务是否结束
	rollbackOnly   bool           //是否只回滚
	transDepth     int            //调用的深度
	sharDBSerevcie ShardDBService //分片服务
}

// DB sql.DB
func (p *Op) DB() *sql.DB {
	return p.pool.db
}

// Pool pool
func (p *Op) Pool() *Pool {
	return p.pool
}

// PoolName name of pool
func (p *Op) PoolName() string {
	return p.pool.name
}

// SetupTableShard use op pool setup entity table shard
func (p *Op) SetupTableShard(entity Entity, ruleName string) error {
	if p.sharDBSerevcie == nil {
		return errors.New("no shard db service")
	}
	poolName, err := p.sharDBSerevcie.setupTableShard(entity, ruleName)
	if err != nil {
		return err
	}
	if poolName != p.PoolName() {
		return fmt.Errorf("op.PoolName %s != entity.PoolName %s", p.PoolName(), poolName)
	}
	return nil
}

func (p *Op) close() {
	p.tx = nil
	p.rollbackOnly = false
	p.transDepth = 0
}

//检查事务的状态
func (p *Op) checkTransStatus() error {
	if p.txDone {
		return sql.ErrTxDone
	}
	if p.tx == nil {
		return NewDBError(nil, "Not begin transaction")
	}
	return nil
}

func (p *Op) incrTransDepth() {
	p.transDepth = p.transDepth + 1
	c.Debugf("p.tranDepth:%v", p.transDepth)
}

func (p *Op) decrTransDepth() error {
	p.transDepth = p.transDepth - 1
	c.Debugf("p.tranDepth:%v", p.transDepth)
	if p.transDepth < 0 {
		return NewDBError(nil, "Too many invoke commit or rollback")
	}
	return nil
}

//结束事务
func (p *Op) finishTrans() error {
	if err := p.checkTransStatus(); err != nil {
		return err
	}
	if err := p.decrTransDepth(); err != nil {
		return err
	}
	if p.transDepth > 0 {
		return nil
	}
	defer p.close()
	p.txDone = true
	if p.rollbackOnly {
		c.Debugf("Rollback")
		return p.tx.Rollback()
	}
	c.Debugf("Commit")
	return p.tx.Commit()
}

// BeginTx 开始事务,支持简单的嵌套调用,如果已经开始了事务,则直接返回成功
func (p *Op) BeginTx() (err error) {
	p.incrTransDepth()
	if p.tx != nil {
		return nil //事务已经开启
	}
	if tx, err := p.DB().Begin(); err == nil {
		p.tx = tx
		p.txDone = false
		return nil
	}
	return err
}

// Commit 提交事务
func (p *Op) Commit() error {
	return p.finishTrans()
}

// Rollback 回滚事务
func (p *Op) Rollback() error {
	p.SetRollbackOnly(true)
	return p.finishTrans()
}

// SetRollbackOnly 设置只回滚
func (p *Op) SetRollbackOnly(rollback bool) {
	p.rollbackOnly = rollback
}

// IsRollbackOnly 是否只回滚
func (p *Op) IsRollbackOnly() bool {
	return p.rollbackOnly
}

// DoInTrans 在事务中执行
{'fraction_non_alphanumeric': 0.14159610880922358, 'fraction_numerical': 0.0021617726535759323, 'mean_word_length': 3.0, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2031565', 'n_tokens_mistral': 2377, 'n_tokens_neox': 2401, 'n_words': 586}