-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
added workflow and execution db entity
- Loading branch information
Showing
11 changed files
with
412 additions
and
53 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| package gormimpl | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/prometheus/client_golang/prometheus" | ||
| "github.com/prometheus/client_golang/prometheus/promauto" | ||
| "gorm.io/gorm" | ||
|
|
||
| interfaces "task/server/repository/interface" | ||
| models "task/server/repository/model/task" | ||
| ) | ||
|
|
||
| var ( | ||
| executionOperations = promauto.NewCounterVec( | ||
| prometheus.CounterOpts{ | ||
| Name: "execution_repository_operations_total", | ||
| Help: "The total number of execution repository operations", | ||
| }, | ||
| []string{"operation", "status"}, | ||
| ) | ||
| executionLatency = promauto.NewHistogramVec( | ||
| prometheus.HistogramOpts{ | ||
| Name: "execution_repository_operation_duration_seconds", | ||
| Help: "Duration of execution repository operations in seconds", | ||
| Buckets: prometheus.DefBuckets, | ||
| }, | ||
| []string{"operation"}, | ||
| ) | ||
| ) | ||
|
|
||
| // ExecutionRepo implements the ExecutionRepo interface using GORM for database operations | ||
| // and River for execution queue management. | ||
| type ExecutionRepo struct { | ||
| db *gorm.DB | ||
| } | ||
|
|
||
| // CreateExecution creates a new execution in the database and enqueues it for processing. | ||
| // It returns the created execution with its assigned ID or an error if the operation fails. | ||
| func (s *ExecutionRepo) CreateExecution(ctx context.Context, execution models.Execution) (models.Execution, error) { | ||
| timer := prometheus.NewTimer(executionLatency.WithLabelValues("create")) | ||
| defer timer.ObserveDuration() | ||
|
|
||
| result := s.db.Create(&execution) | ||
| if result.Error != nil { | ||
| executionOperations.WithLabelValues("create", "error").Inc() | ||
| return models.Execution{}, fmt.Errorf("failed to create execution: %w", result.Error) | ||
| } | ||
|
|
||
| if execution.ID == 0 { | ||
| executionOperations.WithLabelValues("create", "error").Inc() | ||
| return models.Execution{}, fmt.Errorf("failed to get execution ID after creation") | ||
| } | ||
|
|
||
| executionOperations.WithLabelValues("create", "success").Inc() | ||
| return execution, nil | ||
| } | ||
|
|
||
| // GetExecution retrieves a execution from the database by its ID. | ||
| // It returns a pointer to the execution if found, or an error if the execution doesn't exist or if the operation fails. | ||
| func (s *ExecutionRepo) GetExecution(ctx context.Context, executionID uint) (*models.Execution, error) { | ||
| timer := prometheus.NewTimer(executionLatency.WithLabelValues("get")) | ||
| defer timer.ObserveDuration() | ||
|
|
||
| var execution models.Execution | ||
| if err := s.db.First(&execution, executionID).Error; err != nil { | ||
| executionOperations.WithLabelValues("get", "error").Inc() | ||
| return nil, fmt.Errorf("failed to retrieve execution by ID: %w", err) | ||
| } | ||
| executionOperations.WithLabelValues("get", "success").Inc() | ||
| return &execution, nil | ||
| } | ||
|
|
||
| // ListExecutions retrieves a paginated list of executions from the database, filtered by status and type. | ||
| // The 'limit' parameter specifies the maximum number of executions to return, | ||
| // 'offset' determines the starting point for pagination, | ||
| // 'status' allows filtering by execution status, and 'executionType' allows filtering by execution type. | ||
| // It returns a slice of executions and an error if the operation fails. | ||
| func (s *ExecutionRepo) ListExecution(ctx context.Context) ([]models.Execution, error) { | ||
| timer := prometheus.NewTimer(executionLatency.WithLabelValues("list")) | ||
| defer timer.ObserveDuration() | ||
|
|
||
| var executions []models.Execution | ||
|
|
||
| // Execute the query | ||
| if err := s.db.Find(&executions).Error; err != nil { | ||
| executionOperations.WithLabelValues("list", "error").Inc() | ||
| return nil, fmt.Errorf("failed to retrieve executions: %w", err) | ||
| } | ||
|
|
||
| executionOperations.WithLabelValues("list", "success").Inc() | ||
| return executions, nil | ||
| } | ||
|
|
||
| // NewExecutionRepo creates and returns a new instance of ExecutionRepo. | ||
| // It requires a GORM database connection and a River client for execution queue management. | ||
| func NewExecutionRepo(db *gorm.DB) interfaces.ExecutionRepo { | ||
| return &ExecutionRepo{ | ||
| db: db, | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| package gormimpl | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
|
|
||
| "github.com/prometheus/client_golang/prometheus" | ||
| "github.com/prometheus/client_golang/prometheus/promauto" | ||
| "gorm.io/gorm" | ||
|
|
||
| interfaces "task/server/repository/interface" | ||
| models "task/server/repository/model/task" | ||
| ) | ||
|
|
||
| var ( | ||
| workflowOperations = promauto.NewCounterVec( | ||
| prometheus.CounterOpts{ | ||
| Name: "workflow_repository_operations_total", | ||
| Help: "The total number of workflow repository operations", | ||
| }, | ||
| []string{"operation", "status"}, | ||
| ) | ||
| workflowLatency = promauto.NewHistogramVec( | ||
| prometheus.HistogramOpts{ | ||
| Name: "workflow_repository_operation_duration_seconds", | ||
| Help: "Duration of workflow repository operations in seconds", | ||
| Buckets: prometheus.DefBuckets, | ||
| }, | ||
| []string{"operation"}, | ||
| ) | ||
| ) | ||
|
|
||
| // WorkflowRepo implements the WorkflowRepo interface using GORM for database operations | ||
| // and River for workflow queue management. | ||
| type WorkflowRepo struct { | ||
| db *gorm.DB | ||
| } | ||
|
|
||
| // CreateWorkflow creates a new workflow in the database and enqueues it for processing. | ||
| // It returns the created workflow with its assigned ID or an error if the operation fails. | ||
| func (s *WorkflowRepo) CreateWorkflow(ctx context.Context, workflow models.Workflow) (models.Workflow, error) { | ||
| timer := prometheus.NewTimer(workflowLatency.WithLabelValues("create")) | ||
| defer timer.ObserveDuration() | ||
|
|
||
| result := s.db.Create(&workflow) | ||
| if result.Error != nil { | ||
| workflowOperations.WithLabelValues("create", "error").Inc() | ||
| return models.Workflow{}, fmt.Errorf("failed to create workflow: %w", result.Error) | ||
| } | ||
|
|
||
| if workflow.ID == 0 { | ||
| workflowOperations.WithLabelValues("create", "error").Inc() | ||
| return models.Workflow{}, fmt.Errorf("failed to get workflow ID after creation") | ||
| } | ||
|
|
||
| workflowOperations.WithLabelValues("create", "success").Inc() | ||
| return workflow, nil | ||
| } | ||
|
|
||
| // GetWorkflow retrieves a workflow from the database by its ID. | ||
| // It returns a pointer to the workflow if found, or an error if the workflow doesn't exist or if the operation fails. | ||
| func (s *WorkflowRepo) GetWorkflow(ctx context.Context, workflowID uint) (*models.Workflow, error) { | ||
| timer := prometheus.NewTimer(workflowLatency.WithLabelValues("get")) | ||
| defer timer.ObserveDuration() | ||
|
|
||
| var workflow models.Workflow | ||
| if err := s.db.First(&workflow, workflowID).Error; err != nil { | ||
| workflowOperations.WithLabelValues("get", "error").Inc() | ||
| return nil, fmt.Errorf("failed to retrieve workflow by ID: %w", err) | ||
| } | ||
| workflowOperations.WithLabelValues("get", "success").Inc() | ||
| return &workflow, nil | ||
| } | ||
|
|
||
| // ListWorkflows retrieves a paginated list of workflows from the database, filtered by status and type. | ||
| // The 'limit' parameter specifies the maximum number of workflows to return, | ||
| // 'offset' determines the starting point for pagination, | ||
| // 'status' allows filtering by workflow status, and 'workflowType' allows filtering by workflow type. | ||
| // It returns a slice of workflows and an error if the operation fails. | ||
| func (s *WorkflowRepo) ListWorkflow(ctx context.Context) ([]models.Workflow, error) { | ||
| timer := prometheus.NewTimer(workflowLatency.WithLabelValues("list")) | ||
| defer timer.ObserveDuration() | ||
|
|
||
| var workflows []models.Workflow | ||
|
|
||
| // Execute the query | ||
| if err := s.db.Find(&workflows).Error; err != nil { | ||
| workflowOperations.WithLabelValues("list", "error").Inc() | ||
| return nil, fmt.Errorf("failed to retrieve workflows: %w", err) | ||
| } | ||
|
|
||
| workflowOperations.WithLabelValues("list", "success").Inc() | ||
| return workflows, nil | ||
| } | ||
|
|
||
| // NewWorkflowRepo creates and returns a new instance of WorkflowRepo. | ||
| // It requires a GORM database connection and a River client for workflow queue management. | ||
| func NewWorkflowRepo(db *gorm.DB) interfaces.WorkflowRepo { | ||
| return &WorkflowRepo{ | ||
| db: db, | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| package interfaces | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| model "task/server/repository/model/task" | ||
| ) | ||
|
|
||
| // ExecutionRepo defines the interface for the task history repository. | ||
| // It handles operations related to task history management. | ||
| // | ||
| //go:generate mockery --output=../mocks --case=underscore --all --with-expecter | ||
| type ExecutionRepo interface { | ||
| // CreateExecution creates a history entry for a task. | ||
| // It takes a context.Context parameter for handling request-scoped values and deadlines. | ||
| CreateExecution(ctx context.Context, execution model.Execution) (model.Execution, error) | ||
|
|
||
| // GetExecution retrieves the history of a task by its ID. | ||
| // Returns a slice of task history entries, or an error if none found. | ||
| GetExecution(ctx context.Context, taskID uint) (*model.Execution, error) | ||
|
|
||
| ListExecution(ctx context.Context) ([]model.Execution, error) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.