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).
98 lines
2.0 KiB
Go
98 lines
2.0 KiB
Go
package mysql
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
|
|
"github.com/hanzoai/tasks/common/persistence/sql/sqlplugin"
|
|
)
|
|
|
|
const (
|
|
createShardQry = `INSERT INTO
|
|
shards (shard_id, range_id, data, data_encoding) VALUES (?, ?, ?, ?)`
|
|
|
|
getShardQry = `SELECT
|
|
shard_id, range_id, data, data_encoding
|
|
FROM shards WHERE shard_id = ?`
|
|
|
|
updateShardQry = `UPDATE shards
|
|
SET range_id = ?, data = ?, data_encoding = ?
|
|
WHERE shard_id = ?`
|
|
|
|
lockShardQry = `SELECT range_id FROM shards WHERE shard_id = ? FOR UPDATE`
|
|
readLockShardQry = `SELECT range_id FROM shards WHERE shard_id = ? LOCK IN SHARE MODE`
|
|
)
|
|
|
|
// InsertIntoShards inserts one or more rows into shards table
|
|
func (mdb *db) InsertIntoShards(
|
|
ctx context.Context,
|
|
row *sqlplugin.ShardsRow,
|
|
) (sql.Result, error) {
|
|
return mdb.ExecContext(ctx,
|
|
createShardQry,
|
|
row.ShardID,
|
|
row.RangeID,
|
|
row.Data,
|
|
row.DataEncoding,
|
|
)
|
|
}
|
|
|
|
// UpdateShards updates one or more rows into shards table
|
|
func (mdb *db) UpdateShards(
|
|
ctx context.Context,
|
|
row *sqlplugin.ShardsRow,
|
|
) (sql.Result, error) {
|
|
return mdb.ExecContext(ctx,
|
|
updateShardQry,
|
|
row.RangeID,
|
|
row.Data,
|
|
row.DataEncoding,
|
|
row.ShardID,
|
|
)
|
|
}
|
|
|
|
// SelectFromShards reads one or more rows from shards table
|
|
func (mdb *db) SelectFromShards(
|
|
ctx context.Context,
|
|
filter sqlplugin.ShardsFilter,
|
|
) (*sqlplugin.ShardsRow, error) {
|
|
var row sqlplugin.ShardsRow
|
|
err := mdb.GetContext(ctx,
|
|
&row,
|
|
getShardQry,
|
|
filter.ShardID,
|
|
)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &row, err
|
|
}
|
|
|
|
// ReadLockShards acquires a read lock on a single row in shards table
|
|
func (mdb *db) ReadLockShards(
|
|
ctx context.Context,
|
|
filter sqlplugin.ShardsFilter,
|
|
) (int64, error) {
|
|
var rangeID int64
|
|
err := mdb.GetContext(ctx,
|
|
&rangeID,
|
|
readLockShardQry,
|
|
filter.ShardID,
|
|
)
|
|
return rangeID, err
|
|
}
|
|
|
|
// WriteLockShards acquires a write lock on a single row in shards table
|
|
func (mdb *db) WriteLockShards(
|
|
ctx context.Context,
|
|
filter sqlplugin.ShardsFilter,
|
|
) (int64, error) {
|
|
var rangeID int64
|
|
err := mdb.GetContext(ctx,
|
|
&rangeID,
|
|
lockShardQry,
|
|
filter.ShardID,
|
|
)
|
|
return rangeID, err
|
|
}
|