!feat(dtc):renamed Context and added DTC

CHANGELOG.md

@@ -5,10 +5,14 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
-## [1.4.7]
+## [1.5.0]
+### Changed
+- renamed `Context` with `Conn` (#44)
+
 ### Added
 - added `Connector` interface (#43)
 - added nullable `Time` with better json support (#44)
+- added `DistributedTransactions` (#44)
 
 ### Changed
 - renamed `BitBool` with shorter name `Bool` (#44)

context.go → conn.go

@@ -8,7 +8,7 @@
 	"time"
 )
 
-type Context struct {
+type Conn struct {
 	*sql.DB
 	sync.Mutex
 	_ noCopy
@@ -20,11 +20,11 @@
 	Index           int
 }
 
-func (db *Context) Query(query string, args ...any) (*Rows, error) {
+func (db *Conn) Query(query string, args ...any) (*Rows, error) {
 	return db.QueryContext(context.Background(), query, args...)
 }
 
-func (db *Context) QueryBuilder(ctx context.Context, b *Builder) (*Rows, error) {
+func (db *Conn) QueryBuilder(ctx context.Context, b *Builder) (*Rows, error) {
 	query, args, err := b.Build()
 	if err != nil {
 		return nil, err
@@ -33,7 +33,7 @@
 	return db.QueryContext(ctx, query, args...)
 }
 
-func (db *Context) QueryContext(ctx context.Context, query string, args ...any) (*Rows, error) {
+func (db *Conn) QueryContext(ctx context.Context, query string, args ...any) (*Rows, error) {
 	var rows *sql.Rows
 	var stmt *Stmt
 	var err error
@@ -59,11 +59,11 @@
 	return &Rows{Rows: rows, stmt: stmt, query: query}, nil
 }
 
-func (db *Context) QueryRow(query string, args ...any) *Row {
+func (db *Conn) QueryRow(query string, args ...any) *Row {
 	return db.QueryRowContext(context.Background(), query, args...)
 }
 
-func (db *Context) QueryRowBuilder(ctx context.Context, b *Builder) *Row {
+func (db *Conn) QueryRowBuilder(ctx context.Context, b *Builder) *Row {
 	query, args, err := b.Build()
 	if err != nil {
 		return &Row{
@@ -75,7 +75,7 @@
 	return db.QueryRowContext(ctx, query, args...)
 }
 
-func (db *Context) QueryRowContext(ctx context.Context, query string, args ...any) *Row {
+func (db *Conn) QueryRowContext(ctx context.Context, query string, args ...any) *Row {
 	var rows *sql.Rows
 	var stmt *Stmt
 	var err error
@@ -108,11 +108,11 @@
 	}
 }
 
-func (db *Context) Exec(query string, args ...any) (sql.Result, error) {
+func (db *Conn) Exec(query string, args ...any) (sql.Result, error) {
 	return db.ExecContext(context.Background(), query, args...)
 }
 
-func (db *Context) ExecBuilder(ctx context.Context, b *Builder) (sql.Result, error) {
+func (db *Conn) ExecBuilder(ctx context.Context, b *Builder) (sql.Result, error) {
 	query, args, err := b.Build()
 	if err != nil {
 		return nil, err
@@ -121,7 +121,7 @@
 	return db.ExecContext(ctx, query, args...)
 }
 
-func (db *Context) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
+func (db *Conn) ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error) {
 	if len(args) > 0 {
 		stmt, err := db.prepareStmt(ctx, query)
 		if err != nil {
@@ -135,12 +135,12 @@
 	return db.DB.ExecContext(context.Background(), query, args...)
 }
 
-func (db *Context) Begin(opts *sql.TxOptions) (*Tx, error) {
+func (db *Conn) Begin(opts *sql.TxOptions) (*Tx, error) {
 	return db.BeginTx(context.TODO(), opts)
 
 }
 
-func (db *Context) BeginTx(ctx context.Context, opts *sql.TxOptions) (*Tx, error) {
+func (db *Conn) BeginTx(ctx context.Context, opts *sql.TxOptions) (*Tx, error) {
 	tx, err := db.DB.BeginTx(ctx, opts)
 	if err != nil {
 		return nil, err
@@ -149,7 +149,7 @@
 	return &Tx{Tx: tx, stmts: make(map[string]*sql.Stmt)}, nil
 }
 
-func (db *Context) Transaction(ctx context.Context, opts *sql.TxOptions, fn func(ctx context.Context, tx *Tx) error) error {
+func (db *Conn) Transaction(ctx context.Context, opts *sql.TxOptions, fn func(ctx context.Context, tx *Tx) error) error {
 	tx, err := db.BeginTx(ctx, opts)
 	if err != nil {
 		return err

context_stmt.go → conn_stmt.go

@@ -21,7 +21,7 @@ func (s *Stmt) Reuse() {
 	s.isUsing = false
 }
 
-func (db *Context) prepareStmt(ctx context.Context, query string) (*Stmt, error) {
+func (db *Conn) prepareStmt(ctx context.Context, query string) (*Stmt, error) {
 	db.stmtsMutex.Lock()
 	defer db.stmtsMutex.Unlock()
 	s, ok := db.stmts[query]
@@ -48,7 +48,7 @@ func (db *Context) prepareStmt(ctx context.Context, query string) (*Stmt, error)
 	return s, nil
 }
 
-func (db *Context) closeStaleStmt() {
+func (db *Conn) closeStaleStmt() {
 	db.stmtsMutex.Lock()
 	defer db.stmtsMutex.Unlock()
 
@@ -64,7 +64,7 @@ func (db *Context) closeStaleStmt() {
 
 }
 
-func (db *Context) checkIdleStmt() {
+func (db *Conn) checkIdleStmt() {
 	delay := time.NewTicker(db.stmtMaxIdleTime)
 	defer delay.Stop()
 

connector.go

@@ -6,7 +6,7 @@ import (
 )
 
 // Connector represents a database connector that provides methods for executing queries and commands.
-// Context and Tx both implement this interface.
+// Conn and Tx both implement this interface.
 type Connector interface {
 	// Query executes a query that returns multiple rows.
 	// It takes a query string and optional arguments.

db.go

@@ -16,12 +16,12 @@ var (
 
 // DB represents a database connection pool with sharding support.
 type DB struct {
-	*Context
+	*Conn
 	_ noCopy //nolint: unused
 
 	mu   sync.RWMutex
 	dhts map[string]*shardid.DHT
-	dbs  []*Context
+	dbs  []*Conn
 }
 
 // Open creates a new DB instance with the provided database connections.
@@ -31,7 +31,7 @@ func Open(dbs ...*sql.DB) *DB {
 	}
 
 	for i, db := range dbs {
-		ctx := &Context{
+		ctx := &Conn{
 			DB:              db,
 			Index:           i,
 			stmts:           make(map[string]*Stmt),
@@ -41,7 +41,7 @@ func Open(dbs ...*sql.DB) *DB {
 		go ctx.checkIdleStmt()
 	}
 
-	d.Context = d.dbs[0]
+	d.Conn = d.dbs[0]
 
 	return d
 }
@@ -54,7 +54,7 @@ func (db *DB) Add(dbs ...*sql.DB) {
 	n := len(db.dbs)
 
 	for i, d := range dbs {
-		ctx := &Context{
+		ctx := &Conn{
 			DB:              d,
 			Index:           n + i,
 			stmts:           make(map[string]*Stmt),
@@ -66,7 +66,7 @@ func (db *DB) Add(dbs ...*sql.DB) {
 }
 
 // On selects the database context based on the shardid ID.
-func (db *DB) On(id shardid.ID) *Context {
+func (db *DB) On(id shardid.ID) *Conn {
 	db.mu.RLock()
 	defer db.mu.RUnlock()
 
@@ -90,7 +90,7 @@ func (db *DB) GetDHT(name string) *shardid.DHT {
 }
 
 // OnDHT selects the database context based on the DHT (Distributed Hash Table) key.
-func (db *DB) OnDHT(key string, names ...string) (*Context, error) {
+func (db *DB) OnDHT(key string, names ...string) (*Conn, error) {
 	db.mu.RLock()
 	defer db.mu.RUnlock()
 

dtc.go

@@ -0,0 +1,108 @@
+package sqle
+
+import (
+	"context"
+	"database/sql"
+)
+
+// DTC Distributed Transaction Coordinator
+type DTC struct {
+	ctx  context.Context
+	opts *sql.TxOptions
+
+	sessions []*session
+}
+
+// session represents a transaction session.
+type session struct {
+	committed bool
+	conn      *Conn
+	tx        *Tx
+	exec      []func(ctx context.Context, c Connector) error
+	revert    []func(ctx context.Context, c Connector) error
+}
+
+// NewDTC creates a new instance of DTC.
+func NewDTC(ctx context.Context, opts *sql.TxOptions) *DTC {
+	return &DTC{
+		ctx:  ctx,
+		opts: opts,
+	}
+}
+
+// Prepare adds a new transaction session to the DTC.
+func (d *DTC) Prepare(conn *Conn, exec func(ctx context.Context, c Connector) error, revert func(ctx context.Context, c Connector) error) {
+	for _, s := range d.sessions {
+		if s.conn.Index == conn.Index {
+			s.exec = append(s.exec, exec)
+			s.revert = append(s.revert, revert)
+			return
+		}
+	}
+
+	s := &session{
+		committed: false,
+		conn:      conn,
+		exec: []func(ctx context.Context, c Connector) error{
+			exec,
+		},
+		revert: []func(ctx context.Context, c Connector) error{
+			revert,
+		},
+	}
+
+	d.sessions = append(d.sessions, s)
+
+}
+
+// Commit commits all the prepared transactions in the DTC.
+func (d *DTC) Commit() error {
+	for _, s := range d.sessions {
+		tx, err := s.conn.BeginTx(d.ctx, d.opts)
+		if err != nil {
+			return err
+		}
+
+		s.tx = tx
+
+		for _, exec := range s.exec {
+			err = exec(d.ctx, tx)
+			if err != nil {
+				return err
+			}
+		}
+	}
+
+	for _, s := range d.sessions {
+		err := s.tx.Commit()
+		if err != nil {
+			return err
+		}
+
+		s.committed = true
+	}
+
+	return nil
+}
+
+// Rollback rolls back all the prepared transactions in the DTC.
+func (d *DTC) Rollback() []error {
+	var errs []error
+
+	for _, s := range d.sessions {
+		if s.committed {
+			for _, revert := range s.revert {
+				if err := revert(d.ctx, s.conn); err != nil {
+					errs = append(errs, err)
+				}
+			}
+
+		} else {
+			if err := s.tx.Rollback(); err != nil {
+				errs = append(errs, err)
+			}
+		}
+	}
+
+	return errs
+}