mirror of
https://github.com/hanzoai/tasks.git
synced 2026-11-03 03:56:04 +00:00
zeekay
5a601d081b
Full module path rebrand from Temporal to Hanzo Tasks: - go.mod module: github.com/hanzoai/tasks - All internal Go imports updated (1906 files) - cmd/server renamed to cmd/tasksd - Binary name: tasksd (was temporal-server) - Docker images: hanzoai/tasks (was temporalio/server) - README rewritten with Hanzo Tasks branding - LLM.md created with integration notes - Makefile, goreleaser, CI workflows updated - go.temporal.io/sdk and go.temporal.io/api imports preserved (separate repos) Fork of Temporal.io (MIT license).
138 lines
3.3 KiB
Go
138 lines
3.3 KiB
Go
package sql
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"database/sql"
|
|
"encoding/binary"
|
|
"encoding/gob"
|
|
"encoding/json"
|
|
"fmt"
|
|
|
|
"go.temporal.io/api/serviceerror"
|
|
"github.com/hanzoai/tasks/common/log"
|
|
"github.com/hanzoai/tasks/common/log/tag"
|
|
"github.com/hanzoai/tasks/common/persistence"
|
|
"github.com/hanzoai/tasks/common/persistence/serialization"
|
|
"github.com/hanzoai/tasks/common/persistence/sql/sqlplugin"
|
|
)
|
|
|
|
// TODO: Rename all SQL Managers to Stores
|
|
type SqlStore struct {
|
|
DB sqlplugin.DB
|
|
logger log.Logger
|
|
serializer serialization.Serializer
|
|
}
|
|
|
|
func NewSQLStore(db sqlplugin.DB, logger log.Logger, serializer serialization.Serializer) SqlStore {
|
|
return SqlStore{
|
|
DB: db,
|
|
logger: logger,
|
|
serializer: serializer,
|
|
}
|
|
}
|
|
|
|
func (m *SqlStore) GetName() string {
|
|
return m.DB.PluginName()
|
|
}
|
|
|
|
func (m *SqlStore) GetDbName() string {
|
|
return m.DB.DbName()
|
|
}
|
|
|
|
func (m *SqlStore) Close() {
|
|
if m.DB != nil {
|
|
err := m.DB.Close()
|
|
if err != nil {
|
|
m.logger.Error("Error closing SQL database", tag.Error(err))
|
|
}
|
|
}
|
|
}
|
|
|
|
func (m *SqlStore) txExecute(ctx context.Context, operation string, f func(tx sqlplugin.Tx) error) error {
|
|
tx, err := m.DB.BeginTx(ctx)
|
|
if err != nil {
|
|
return serviceerror.NewUnavailablef("%s failed. Failed to start transaction. Error: %v", operation, err)
|
|
}
|
|
err = f(tx)
|
|
if err != nil {
|
|
rollBackErr := tx.Rollback()
|
|
if rollBackErr != nil {
|
|
m.logger.Error("transaction rollback error", tag.Error(rollBackErr))
|
|
}
|
|
|
|
switch err.(type) {
|
|
case *persistence.ConditionFailedError,
|
|
*persistence.CurrentWorkflowConditionFailedError,
|
|
*persistence.WorkflowConditionFailedError,
|
|
*serviceerror.NamespaceAlreadyExists,
|
|
*persistence.ShardOwnershipLostError,
|
|
*serviceerror.Unavailable,
|
|
*serviceerror.NotFound:
|
|
return err
|
|
default:
|
|
return serviceerror.NewUnavailablef("%v: %v", operation, err)
|
|
}
|
|
}
|
|
if err := tx.Commit(); err != nil {
|
|
return serviceerror.NewUnavailablef("%s operation failed. Failed to commit transaction. Error: %v", operation, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func gobSerialize(x any) ([]byte, error) {
|
|
b := bytes.Buffer{}
|
|
e := gob.NewEncoder(&b)
|
|
err := e.Encode(x)
|
|
if err != nil {
|
|
return nil, serviceerror.NewInternalf("Error in serialization: %v", err)
|
|
}
|
|
return b.Bytes(), nil
|
|
}
|
|
|
|
func gobDeserialize(a []byte, x any) error {
|
|
b := bytes.NewBuffer(a)
|
|
d := gob.NewDecoder(b)
|
|
err := d.Decode(x)
|
|
if err != nil {
|
|
return serviceerror.NewInternalf("Error in deserialization: %v", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func serializePageToken(offset int64) []byte {
|
|
b := make([]byte, 8)
|
|
binary.LittleEndian.PutUint64(b, uint64(offset))
|
|
return b
|
|
}
|
|
|
|
func deserializePageToken(payload []byte) (int64, error) {
|
|
if len(payload) != 8 {
|
|
return 0, fmt.Errorf("invalid token of %v length", len(payload))
|
|
}
|
|
return int64(binary.LittleEndian.Uint64(payload)), nil
|
|
}
|
|
|
|
func serializePageTokenJson[T any](token *T) ([]byte, error) {
|
|
return json.Marshal(token)
|
|
}
|
|
|
|
func deserializePageTokenJson[T any](payload []byte) (*T, error) {
|
|
var token T
|
|
if err := json.Unmarshal(payload, &token); err != nil {
|
|
return nil, err
|
|
}
|
|
return &token, nil
|
|
}
|
|
|
|
func convertCommonErrors(
|
|
operation string,
|
|
err error,
|
|
) error {
|
|
if err == sql.ErrNoRows {
|
|
return serviceerror.NewNotFoundf("%v failed. Error: %v ", operation, err)
|
|
}
|
|
|
|
return serviceerror.NewUnavailablef("%v operation failed. Error: %v", operation, err)
|
|
}
|