MiFaZhan
117 lines
4.2 KiB
Go
117 lines
4.2 KiB
Go
package contract
|
|
|
|
import (
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
)
|
|
|
|
const (
|
|
DefaultFlowPrefix = "flow:"
|
|
DefaultSimStatusPrefix = "sim-status:"
|
|
)
|
|
|
|
// Keys is a naming contract only. It does not create indexes or storage.
|
|
// Existing run names are byte-for-byte compatible with zset.Store. Global
|
|
// run/archive indexes are outside run hash slots and cannot join run-local Lua.
|
|
type Keys struct{ prefix string }
|
|
|
|
func NewKeys(prefix string) (Keys, error) {
|
|
if strings.TrimSpace(prefix) == "" {
|
|
prefix = DefaultFlowPrefix
|
|
}
|
|
if strings.ContainsAny(prefix, "{}") {
|
|
return Keys{}, invalid("prefix must not contain Redis hash tags")
|
|
}
|
|
return Keys{prefix: prefix}, nil
|
|
}
|
|
|
|
// The zero value uses the legacy flow prefix, just like NewKeys("").
|
|
func (k Keys) Prefix() string {
|
|
if k.prefix == "" {
|
|
return DefaultFlowPrefix
|
|
}
|
|
return k.prefix
|
|
}
|
|
|
|
func (k Keys) runKey(runID, suffix string) (string, error) {
|
|
if err := ValidateID(runID); err != nil {
|
|
return "", err
|
|
}
|
|
return k.Prefix() + "{" + runID + "}:" + suffix, nil
|
|
}
|
|
|
|
func (k Keys) RunMetaKey(runID string) (string, error) { return k.runKey(runID, "meta") }
|
|
func (k Keys) RunChannelsKey(runID string) (string, error) { return k.runKey(runID, "channels") }
|
|
func (k Keys) TaskStateKey(runID string) (string, error) { return k.runKey(runID, "task-state") }
|
|
func (k Keys) FailedKey(runID string) (string, error) { return k.runKey(runID, "failed") }
|
|
func (k Keys) ArchiveStateKey(runID string) (string, error) { return k.runKey(runID, "archive") }
|
|
|
|
func (k Keys) roundKey(runID string, round int, suffix string) (string, error) {
|
|
if round < 1 {
|
|
return "", invalid("round must be positive")
|
|
}
|
|
return k.runKey(runID, fmt.Sprintf("round:%d:%s", round, suffix))
|
|
}
|
|
|
|
func (k Keys) RoundStatsKey(runID string, round int) (string, error) {
|
|
return k.roundKey(runID, round, "stats")
|
|
}
|
|
|
|
func (k Keys) roundChannelKey(runID string, round int, kind, chanKey string) (string, error) {
|
|
if strings.TrimSpace(chanKey) == "" {
|
|
return "", invalid("chanKey is empty")
|
|
}
|
|
return k.roundKey(runID, round, kind+":"+chanKey)
|
|
}
|
|
|
|
func (k Keys) RoundActiveKey(runID string, round int, chanKey string) (string, error) {
|
|
return k.roundChannelKey(runID, round, "active", chanKey)
|
|
}
|
|
|
|
func (k Keys) RoundWaitingKey(runID string, round int, chanKey string) (string, error) {
|
|
return k.roundChannelKey(runID, round, "waiting", chanKey)
|
|
}
|
|
|
|
func (k Keys) RoundLeaseKey(runID string, round int, chanKey string) (string, error) {
|
|
return k.roundChannelKey(runID, round, "lease", chanKey)
|
|
}
|
|
|
|
func (k Keys) FailedRunsKey() string { return k.Prefix() + "failed:runs" }
|
|
func (k Keys) ActiveRunsKey() string { return k.Prefix() + "runs:active" }
|
|
func (k Keys) AllRunsKey() string { return k.Prefix() + "runs:all" }
|
|
func (k Keys) PendingArchivesKey() string { return k.Prefix() + "archive:pending" }
|
|
|
|
// Replay action state, leases, idempotency reservations and pending index share
|
|
// a dedicated slot, allowing atomic acceptance/discovery. Source/target run
|
|
// writes remain in their own slots and require a recoverable replay protocol.
|
|
func (k Keys) replayKey(actionID, suffix string) (string, error) {
|
|
if err := ValidateID(actionID); err != nil {
|
|
return "", err
|
|
}
|
|
return k.Prefix() + "{replay}:action:" + actionID + ":" + suffix, nil
|
|
}
|
|
|
|
func (k Keys) ReplayActionKey(actionID string) (string, error) { return k.replayKey(actionID, "meta") }
|
|
func (k Keys) ReplayLeaseKey(actionID string) (string, error) { return k.replayKey(actionID, "lease") }
|
|
func (k Keys) PendingReplayActionsKey() string { return k.Prefix() + "{replay}:pending" }
|
|
|
|
// Hash the complete tuple so neither delimiter collisions nor arbitrary
|
|
// Idempotency-Key characters can change the Redis key or hash slot.
|
|
func (k Keys) ReplayIdempotencyKey(sourceRunID, idempotencyKey string) (string, error) {
|
|
if err := ValidateID(sourceRunID); err != nil {
|
|
return "", err
|
|
}
|
|
if err := ValidateIdempotencyKey(idempotencyKey); err != nil {
|
|
return "", err
|
|
}
|
|
data, err := json.Marshal([2]string{sourceRunID, idempotencyKey})
|
|
if err != nil {
|
|
return "", invalid("cannot encode idempotency scope: %v", err)
|
|
}
|
|
digest := sha256.Sum256(data)
|
|
return k.Prefix() + "{replay}:idempotency:" + hex.EncodeToString(digest[:]), nil
|
|
}
|