floxy/store_interface.go at main · floxy-project/floxy · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
package floxy

import (
	"context"
	"encoding/json"
	"time"
)

type Store interface {
	SetAgingEnabled(enabled bool)
	SetAgingRate(rate float64)
	SaveWorkflowDefinition(ctx context.Context, def *WorkflowDefinition) error
	GetWorkflowDefinition(ctx context.Context, id string) (*WorkflowDefinition, error)
	CreateInstance(
		ctx context.Context,
		workflowID string,
		input json.RawMessage,
	) (*WorkflowInstance, error)
	UpdateInstanceStatus(
		ctx context.Context,
		instanceID int64,
		status WorkflowStatus,
		output json.RawMessage,
		errMsg *string,
	) error
	GetInstance(ctx context.Context, instanceID int64) (*WorkflowInstance, error)
	CreateStep(ctx context.Context, step *WorkflowStep) error
	UpdateStep(
		ctx context.Context,
		stepID int64,
		status StepStatus,
		output json.RawMessage,
		errMsg *string,
	) error
	GetStepsByInstance(ctx context.Context, instanceID int64) ([]WorkflowStep, error)
	EnqueueStep(
		ctx context.Context,
		instanceID int64,
		stepID *int64,
		priority Priority,
		delay time.Duration,
	) error
	UpdateStepCompensationRetry(
		ctx context.Context,
		stepID int64,
		retryCount int,
		status StepStatus,
	) error
	DequeueStep(ctx context.Context, workerID string) (*QueueItem, error)
	RemoveFromQueue(ctx context.Context, queueID int64) error
	ReleaseQueueItem(ctx context.Context, queueID int64) error
	RescheduleAndReleaseQueueItem(ctx context.Context, queueID int64, delay time.Duration) error
	LogEvent(
		ctx context.Context,
		instanceID int64,
		stepID *int64,
		eventType string,
		payload any,
	) error
	CreateJoinState(
		ctx context.Context,
		instanceID int64,
		joinStepName string,
		waitingFor []string,
		strategy JoinStrategy,
	) error
	UpdateJoinState(
		ctx context.Context,
		instanceID int64,
		joinStepName, completedStep string,
		success bool,
	) (bool, error)
	GetJoinState(ctx context.Context, instanceID int64, joinStepName string) (*JoinState, error)
	AddToJoinWaitFor(ctx context.Context, instanceID int64, joinStepName, stepToAdd string) error
	ReplaceInJoinWaitFor(ctx context.Context, instanceID int64, joinStepName, virtualStep, realStep string) error
	GetSummaryStats(ctx context.Context) (*SummaryStats, error)
	GetActiveInstances(ctx context.Context) ([]ActiveWorkflowInstance, error)
	GetWorkflowDefinitions(ctx context.Context) ([]WorkflowDefinition, error)
	GetWorkflowInstances(ctx context.Context, workflowID string) ([]WorkflowInstance, error)
	GetAllWorkflowInstances(ctx context.Context) ([]WorkflowInstance, error)
	GetWorkflowInstancesPaginated(ctx context.Context, workflowID string, offset int, limit int) ([]WorkflowInstance, int64, error)
	GetAllWorkflowInstancesPaginated(ctx context.Context, offset int, limit int) ([]WorkflowInstance, int64, error)
	GetWorkflowSteps(ctx context.Context, instanceID int64) ([]WorkflowStep, error)
	GetWorkflowEvents(ctx context.Context, instanceID int64) ([]WorkflowEvent, error)
	GetWorkflowStats(ctx context.Context) ([]WorkflowStats, error)
	GetActiveStepsForUpdate(ctx context.Context, instanceID int64) ([]WorkflowStep, error)
	CreateCancelRequest(ctx context.Context, req *WorkflowCancelRequest) error
	GetCancelRequest(ctx context.Context, instanceID int64) (*WorkflowCancelRequest, error)
	DeleteCancelRequest(ctx context.Context, instanceID int64) error

	// Human decision methods
	CreateHumanDecision(ctx context.Context, decision *HumanDecisionRecord) error
	GetHumanDecision(ctx context.Context, stepID int64) (*HumanDecisionRecord, error)
	UpdateStepStatus(ctx context.Context, stepID int64, status StepStatus) error
	GetStepByID(ctx context.Context, stepID int64) (*WorkflowStep, error)
	GetHumanDecisionStepByInstanceID(ctx context.Context, instanceID int64) (*WorkflowStep, error)

	// DLQ methods
	CreateDeadLetterRecord(ctx context.Context, rec *DeadLetterRecord) error
	RequeueDeadLetter(
		ctx context.Context,
		dlqID int64,
		newInput *json.RawMessage,
	) error
	ListDeadLetters(ctx context.Context, offset int, limit int) ([]DeadLetterRecord, int64, error)
	GetDeadLetterByID(ctx context.Context, id int64) (*DeadLetterRecord, error)
	PauseActiveStepsAndClearQueue(ctx context.Context, instanceID int64) error

	// Cleanup methods
	CleanupOldWorkflows(ctx context.Context, daysToKeep int) (int64, error)
}