You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by di...@apache.org on 2026/08/21 13:24:17 UTC
(airavata) branch refactor-single-module updated: Refactorign process models
This is an automated email from the ASF dual-hosted git repository.
DImuthuUpe pushed a commit to branch refactor-single-module
in repository https://gitbox.apache.org/repos/asf/airavata.git
The following commit(s) were added to refs/heads/refactor-single-module by this push:
new f40ac90d56 Refactorign process models
f40ac90d56 is described below
commit f40ac90d5629fbd75fd8437fd94a7279fadc77b8
Author: DImuthuUpe <di...@gmail.com>
AuthorDate: Fri Aug 21 09:24:10 2026 -0400
Refactorign process models
---
api/process/controller/process.go | 93 -------
api/process/controller/status.go | 46 ----
api/process/controller/task.go | 130 ---------
api/process/dto/batch_job_process.go | 60 ----
api/process/dto/batch_job_process_status.go | 36 ---
api/process/dto/task.go | 302 ---------------------
.../model/{batch_job_process.go => batch.go} | 31 +--
api/process/model/batch_job_process_status.go | 48 ----
api/process/model/process.go | 121 +++++++++
api/process/model/task.go | 25 +-
api/process/repository/process.go | 72 -----
api/process/repository/status.go | 48 ----
api/process/repository/task.go | 94 -------
api/process/service/process.go | 197 --------------
api/process/service/status.go | 105 -------
api/process/service/task.go | 209 --------------
16 files changed, 137 insertions(+), 1480 deletions(-)
diff --git a/api/process/controller/process.go b/api/process/controller/process.go
deleted file mode 100644
index e594a7adb9..0000000000
--- a/api/process/controller/process.go
+++ /dev/null
@@ -1,93 +0,0 @@
-// Package controller serves the batch job process and process status routes.
-package controller
-
-import (
- "net/http"
-
- "github.com/apache/airavata/internal/httpx"
-
- dto "github.com/apache/airavata/api/process/dto"
- "github.com/apache/airavata/api/process/service"
-)
-
-// ProcessController serves /api/v1/batch-job-processes.
-type ProcessController struct{ svc *service.ProcessService }
-
-// NewProcessController returns a handler delegating to svc.
-func NewProcessController(svc *service.ProcessService) *ProcessController {
- return &ProcessController{svc: svc}
-}
-
-// Register mounts the batch job process routes.
-func (h *ProcessController) Register(mux *http.ServeMux) {
- mux.HandleFunc("GET /api/v1/batch-job-processes", h.list)
- mux.HandleFunc("POST /api/v1/batch-job-processes", h.create)
- mux.HandleFunc("GET /api/v1/batch-job-processes/{processId}", h.get)
- mux.HandleFunc("PUT /api/v1/batch-job-processes/{processId}", h.update)
- mux.HandleFunc("DELETE /api/v1/batch-job-processes/{processId}", h.delete)
-}
-
-// list returns every process, or only those of a deployment when deploymentId is
-// supplied. The two branches have different authorisation: the unfiltered listing is
-// admin only, while the per-deployment one is open.
-func (h *ProcessController) list(w http.ResponseWriter, r *http.Request) {
- var (
- found []dto.Response
- err error
- )
- if deploymentID := r.URL.Query().Get("deploymentId"); deploymentID != "" {
- found, err = h.svc.ListByDeployment(r.Context(), deploymentID)
- } else {
- found, err = h.svc.List(r.Context())
- }
- if err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- httpx.WriteJSON(w, http.StatusOK, found)
-}
-
-func (h *ProcessController) get(w http.ResponseWriter, r *http.Request) {
- proc, err := h.svc.Get(r.Context(), r.PathValue("processId"))
- if err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- httpx.WriteJSON(w, http.StatusOK, proc)
-}
-
-func (h *ProcessController) create(w http.ResponseWriter, r *http.Request) {
- var req dto.Request
- if err := httpx.Bind(r, &req); err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- proc, err := h.svc.Create(r.Context(), &req)
- if err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- httpx.WriteJSON(w, http.StatusCreated, proc)
-}
-
-func (h *ProcessController) update(w http.ResponseWriter, r *http.Request) {
- var req dto.Request
- if err := httpx.Bind(r, &req); err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- proc, err := h.svc.Update(r.Context(), r.PathValue("processId"), &req)
- if err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- httpx.WriteJSON(w, http.StatusOK, proc)
-}
-
-func (h *ProcessController) delete(w http.ResponseWriter, r *http.Request) {
- if err := h.svc.Delete(r.Context(), r.PathValue("processId")); err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- httpx.WriteJSON(w, http.StatusNoContent, nil)
-}
diff --git a/api/process/controller/status.go b/api/process/controller/status.go
deleted file mode 100644
index f24a9a54d0..0000000000
--- a/api/process/controller/status.go
+++ /dev/null
@@ -1,46 +0,0 @@
-package controller
-
-import (
- "net/http"
-
- "github.com/apache/airavata/internal/httpx"
-
- "github.com/apache/airavata/api/process/service"
-)
-
-// StatusController serves read-only access to a process's status history, nested
-// under /api/v1/batch-job-processes/{processId}.
-//
-// There is deliberately no POST or PUT: statuses are recorded internally by
-// service.StatusService — from process submission and from whatever submits and monitors the
-// actual job — never accepted as a request body from a client.
-type StatusController struct{ svc *service.StatusService }
-
-// NewStatusController returns a handler delegating to svc.
-func NewStatusController(svc *service.StatusService) *StatusController {
- return &StatusController{svc: svc}
-}
-
-// Register mounts the read-only status routes.
-func (h *StatusController) Register(mux *http.ServeMux) {
- mux.HandleFunc("GET /api/v1/batch-job-processes/{processId}/statuses", h.list)
- mux.HandleFunc("GET /api/v1/batch-job-processes/{processId}/statuses/{statusId}", h.get)
-}
-
-func (h *StatusController) list(w http.ResponseWriter, r *http.Request) {
- statuses, err := h.svc.ListForProcess(r.Context(), r.PathValue("processId"))
- if err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- httpx.WriteJSON(w, http.StatusOK, statuses)
-}
-
-func (h *StatusController) get(w http.ResponseWriter, r *http.Request) {
- status, err := h.svc.Get(r.Context(), r.PathValue("processId"), r.PathValue("statusId"))
- if err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- httpx.WriteJSON(w, http.StatusOK, status)
-}
diff --git a/api/process/controller/task.go b/api/process/controller/task.go
deleted file mode 100644
index abe3aeffa5..0000000000
--- a/api/process/controller/task.go
+++ /dev/null
@@ -1,130 +0,0 @@
-package controller
-
-import (
- "net/http"
-
- "github.com/apache/airavata/internal/httpx"
-
- dto "github.com/apache/airavata/api/process/dto"
- model "github.com/apache/airavata/api/process/model"
- "github.com/apache/airavata/api/process/service"
-)
-
-// TaskController serves one kind of task nested under
-// /api/v1/batch-job-processes/{processId}.
-//
-// The four kinds are the same five routes over a different payload, so the handlers
-// are written once and instantiated per kind. What differs is supplied at
-// construction: the collection's path segment and how a row is rendered.
-type TaskController[T any, Req any, Res any] struct {
- svc *service.TaskService[T, Req]
- path string
-
- render func(*T) Res
- renderAll func([]T) []Res
-}
-
-// Register mounts the task routes for this kind.
-func (h *TaskController[T, Req, Res]) Register(mux *http.ServeMux) {
- base := "/api/v1/batch-job-processes/{processId}/" + h.path
-
- mux.HandleFunc("GET "+base, h.list)
- mux.HandleFunc("POST "+base, h.create)
- mux.HandleFunc("GET "+base+"/{taskId}", h.get)
- mux.HandleFunc("PUT "+base+"/{taskId}", h.update)
- mux.HandleFunc("DELETE "+base+"/{taskId}", h.delete)
-}
-
-func (h *TaskController[T, Req, Res]) list(w http.ResponseWriter, r *http.Request) {
- tasks, err := h.svc.ListForProcess(r.Context(), r.PathValue("processId"))
- if err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- httpx.WriteJSON(w, http.StatusOK, h.renderAll(tasks))
-}
-
-func (h *TaskController[T, Req, Res]) get(w http.ResponseWriter, r *http.Request) {
- task, err := h.svc.Get(r.Context(), r.PathValue("processId"), r.PathValue("taskId"))
- if err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- httpx.WriteJSON(w, http.StatusOK, h.render(task))
-}
-
-func (h *TaskController[T, Req, Res]) create(w http.ResponseWriter, r *http.Request) {
- var req Req
- if err := httpx.Bind(r, &req); err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- task, err := h.svc.Create(r.Context(), r.PathValue("processId"), &req)
- if err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- httpx.WriteJSON(w, http.StatusCreated, h.render(task))
-}
-
-func (h *TaskController[T, Req, Res]) update(w http.ResponseWriter, r *http.Request) {
- var req Req
- if err := httpx.Bind(r, &req); err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- task, err := h.svc.Update(r.Context(), r.PathValue("processId"), r.PathValue("taskId"), &req)
- if err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- httpx.WriteJSON(w, http.StatusOK, h.render(task))
-}
-
-func (h *TaskController[T, Req, Res]) delete(w http.ResponseWriter, r *http.Request) {
- if err := h.svc.Delete(r.Context(), r.PathValue("processId"), r.PathValue("taskId")); err != nil {
- httpx.WriteError(w, r, err)
- return
- }
- httpx.WriteJSON(w, http.StatusNoContent, nil)
-}
-
-// NewDataStagingTaskController serves .../data-staging-tasks.
-func NewDataStagingTaskController(svc *service.DataStagingTaskService) *TaskController[model.DataStagingTask, dto.DataStagingTaskRequest, dto.DataStagingTaskResponse] {
- return &TaskController[model.DataStagingTask, dto.DataStagingTaskRequest, dto.DataStagingTaskResponse]{
- svc: svc,
- path: "data-staging-tasks",
- render: dto.ToDataStagingTaskResponse,
- renderAll: dto.ToDataStagingTaskResponses,
- }
-}
-
-// NewJobSubmissionTaskController serves .../job-submission-tasks.
-func NewJobSubmissionTaskController(svc *service.JobSubmissionTaskService) *TaskController[model.JobSubmissionTask, dto.JobSubmissionTaskRequest, dto.JobSubmissionTaskResponse] {
- return &TaskController[model.JobSubmissionTask, dto.JobSubmissionTaskRequest, dto.JobSubmissionTaskResponse]{
- svc: svc,
- path: "job-submission-tasks",
- render: dto.ToJobSubmissionTaskResponse,
- renderAll: dto.ToJobSubmissionTaskResponses,
- }
-}
-
-// NewJobMonitoringTaskController serves .../job-monitoring-tasks.
-func NewJobMonitoringTaskController(svc *service.JobMonitoringTaskService) *TaskController[model.JobMonitoringTask, dto.JobMonitoringTaskRequest, dto.JobMonitoringTaskResponse] {
- return &TaskController[model.JobMonitoringTask, dto.JobMonitoringTaskRequest, dto.JobMonitoringTaskResponse]{
- svc: svc,
- path: "job-monitoring-tasks",
- render: dto.ToJobMonitoringTaskResponse,
- renderAll: dto.ToJobMonitoringTaskResponses,
- }
-}
-
-// NewInteractiveCommandTaskController serves .../interactive-command-tasks.
-func NewInteractiveCommandTaskController(svc *service.InteractiveCommandTaskService) *TaskController[model.InteractiveCommandTask, dto.InteractiveCommandTaskRequest, dto.InteractiveCommandTaskResponse] {
- return &TaskController[model.InteractiveCommandTask, dto.InteractiveCommandTaskRequest, dto.InteractiveCommandTaskResponse]{
- svc: svc,
- path: "interactive-command-tasks",
- render: dto.ToInteractiveCommandTaskResponse,
- renderAll: dto.ToInteractiveCommandTaskResponses,
- }
-}
diff --git a/api/process/dto/batch_job_process.go b/api/process/dto/batch_job_process.go
deleted file mode 100644
index f257ba8b2b..0000000000
--- a/api/process/dto/batch_job_process.go
+++ /dev/null
@@ -1,60 +0,0 @@
-package dto
-
-import (
- "github.com/apache/airavata/internal/httpx"
-
- applicationdto "github.com/apache/airavata/api/application/dto"
- model "github.com/apache/airavata/api/process/model"
-)
-
-// Request is the create/update payload for a process.
-//
-// The resource request is carried here rather than copied from the deployment's
-// default, which is what lets a caller ask for different resources for a particular
-// run. There is no owner field: ownership comes from the access token.
-//
-// Java: org.apache.airavata.process.dto.BatchJobProcessRequestDto
-type Request struct {
- DeploymentID string `json:"deploymentId"`
- BatchJobConfig *applicationdto.BatchJobConfigRequest `json:"batchJobConfig"`
-}
-
-// Validate implements httpx.Validator.
-func (r *Request) Validate() []httpx.FieldError {
- var c httpx.Constraints
- c.NotBlank("deploymentId", "Deployment id cannot be blank", r.DeploymentID)
- if r.BatchJobConfig == nil {
- c.Add("batchJobConfig", "Batch job config cannot be null")
- } else {
- c.Nested("batchJobConfig", r.BatchJobConfig)
- }
- return c.Fields()
-}
-
-// Response is the read model for a process. The nested config is a snapshot of what
-// this run actually asked for.
-//
-// Java: org.apache.airavata.process.dto.BatchJobProcessResponseDto
-type Response struct {
- ProcessID string `json:"processId"`
- DeploymentID *string `json:"deploymentId"`
- UserID *string `json:"userId"`
- BatchJobConfig *applicationdto.BatchJobConfigResponse `json:"batchJobConfig"`
-}
-
-func ToResponse(p *model.BatchJobProcess) Response {
- return Response{
- ProcessID: p.ID,
- DeploymentID: p.DeploymentID,
- UserID: p.OwnerID,
- BatchJobConfig: applicationdto.ToBatchJobConfigResponse(p.BatchJobConfig),
- }
-}
-
-func ToResponses(in []model.BatchJobProcess) []Response {
- out := make([]Response, 0, len(in))
- for i := range in {
- out = append(out, ToResponse(&in[i]))
- }
- return out
-}
diff --git a/api/process/dto/batch_job_process_status.go b/api/process/dto/batch_job_process_status.go
deleted file mode 100644
index 77fef686ee..0000000000
--- a/api/process/dto/batch_job_process_status.go
+++ /dev/null
@@ -1,36 +0,0 @@
-package dto
-
-import (
- model "github.com/apache/airavata/api/process/model"
-)
-
-// StatusResponse is the read model for one recorded process status.
-//
-// There is deliberately no request DTO alongside it: statuses are never created or
-// updated from a client request body, only recorded internally by StatusService and
-// read back through it.
-type StatusResponse struct {
- ProcessStatusID string `json:"processStatusId"`
- ProcessID *string `json:"processId"`
- Status *model.BatchProcessStatusType `json:"status"`
- Log *string `json:"log"`
- Timestamp *int64 `json:"timestamp"`
-}
-
-func ToStatusResponse(s *model.BatchJobProcessStatus) StatusResponse {
- return StatusResponse{
- ProcessStatusID: s.ID,
- ProcessID: s.ProcessID,
- Status: s.Status,
- Log: s.Log,
- Timestamp: s.Timestamp,
- }
-}
-
-func ToStatusResponses(in []model.BatchJobProcessStatus) []StatusResponse {
- out := make([]StatusResponse, 0, len(in))
- for i := range in {
- out = append(out, ToStatusResponse(&in[i]))
- }
- return out
-}
diff --git a/api/process/dto/task.go b/api/process/dto/task.go
deleted file mode 100644
index 5509812268..0000000000
--- a/api/process/dto/task.go
+++ /dev/null
@@ -1,302 +0,0 @@
-package dto
-
-import (
- "github.com/apache/airavata/internal/httpx"
-
- model "github.com/apache/airavata/api/process/model"
-)
-
-// The four task kinds share an identity — a task id, the process it belongs to, an
-// on-failure action, a retry count and an execution order — and differ only in what
-// they carry. These DTOs mirror that: the shared constraints are validated by one
-// helper, and each kind adds its own.
-//
-// None of them carries processId or processType. Both come from the path, so a task
-// cannot be moved to a different process by editing it.
-
-// taskCommon validates the fields every task kind shares.
-func taskCommon(c *httpx.Constraints, onFailure *model.OnFailureAction, retryCount, taskOrder *int) {
- if onFailure != nil && !onFailure.Valid() {
- c.Add("onFailure", "On failure action must be one of RETRY, SKIP, EXIT")
- }
- if retryCount != nil && *retryCount < 0 {
- c.Add("retryCount", "Retry count cannot be negative")
- }
- if taskOrder != nil && *taskOrder < 0 {
- c.Add("taskOrder", "Task order cannot be negative")
- }
-}
-
-func validateStorageType(c *httpx.Constraints, field string, t *model.DataStorageType) {
- if t != nil && !t.Valid() {
- c.Add(field, "Data storage type must be one of SCP, S3")
- }
-}
-
-// DataStagingTaskRequest is the create/update payload for a staging task.
-type DataStagingTaskRequest struct {
- SourceDataStorageID *string `json:"sourceDataStorageId"`
- SourceCredentialID *string `json:"sourceCredentialId"`
- SourceDataStorageType *model.DataStorageType `json:"sourceDataStorageType"`
-
- DestinationDataStorageID *string `json:"destinationDataStorageId"`
- DestinationCredentialID *string `json:"destinationCredentialId"`
- DestinationDataStorageType *model.DataStorageType `json:"destinationDataStorageType"`
-
- SourcePath *string `json:"sourcePath"`
- DestinationPath *string `json:"destinationPath"`
-
- OnFailure *model.OnFailureAction `json:"onFailure"`
- RetryCount *int `json:"retryCount"`
- TaskOrder *int `json:"taskOrder"`
-}
-
-// Validate implements httpx.Validator.
-//
-// Both paths are required: a staging task that names neither where data comes from nor
-// where it goes describes no transfer at all.
-func (r *DataStagingTaskRequest) Validate() []httpx.FieldError {
- var c httpx.Constraints
- c.NotBlankPtr("sourcePath", "Source path cannot be blank", r.SourcePath)
- c.NotBlankPtr("destinationPath", "Destination path cannot be blank", r.DestinationPath)
- validateStorageType(&c, "sourceDataStorageType", r.SourceDataStorageType)
- validateStorageType(&c, "destinationDataStorageType", r.DestinationDataStorageType)
- taskCommon(&c, r.OnFailure, r.RetryCount, r.TaskOrder)
- return c.Fields()
-}
-
-func ApplyDataStagingTaskRequest(dst *model.DataStagingTask, src *DataStagingTaskRequest) {
- dst.SourceDataStorageID = src.SourceDataStorageID
- dst.SourceCredentialID = src.SourceCredentialID
- dst.SourceDataStorageType = src.SourceDataStorageType
- dst.DestinationDataStorageID = src.DestinationDataStorageID
- dst.DestinationCredentialID = src.DestinationCredentialID
- dst.DestinationDataStorageType = src.DestinationDataStorageType
- dst.SourcePath = src.SourcePath
- dst.DestinationPath = src.DestinationPath
- dst.OnFailure = src.OnFailure
- dst.RetryCount = src.RetryCount
- dst.TaskOrder = src.TaskOrder
-}
-
-// DataStagingTaskResponse is the read model for a staging task.
-type DataStagingTaskResponse struct {
- TaskID string `json:"taskId"`
- ProcessID *string `json:"processId"`
- ProcessType *string `json:"processType"`
-
- SourceDataStorageID *string `json:"sourceDataStorageId"`
- SourceCredentialID *string `json:"sourceCredentialId"`
- SourceDataStorageType *model.DataStorageType `json:"sourceDataStorageType"`
-
- DestinationDataStorageID *string `json:"destinationDataStorageId"`
- DestinationCredentialID *string `json:"destinationCredentialId"`
- DestinationDataStorageType *model.DataStorageType `json:"destinationDataStorageType"`
-
- SourcePath *string `json:"sourcePath"`
- DestinationPath *string `json:"destinationPath"`
-
- OnFailure *model.OnFailureAction `json:"onFailure"`
- RetryCount *int `json:"retryCount"`
- TaskOrder *int `json:"taskOrder"`
-}
-
-func ToDataStagingTaskResponse(t *model.DataStagingTask) DataStagingTaskResponse {
- return DataStagingTaskResponse{
- TaskID: t.ID,
- ProcessID: t.ProcessID,
- ProcessType: t.ProcessType,
- SourceDataStorageID: t.SourceDataStorageID,
- SourceCredentialID: t.SourceCredentialID,
- SourceDataStorageType: t.SourceDataStorageType,
- DestinationDataStorageID: t.DestinationDataStorageID,
- DestinationCredentialID: t.DestinationCredentialID,
- DestinationDataStorageType: t.DestinationDataStorageType,
- SourcePath: t.SourcePath,
- DestinationPath: t.DestinationPath,
- OnFailure: t.OnFailure,
- RetryCount: t.RetryCount,
- TaskOrder: t.TaskOrder,
- }
-}
-
-func ToDataStagingTaskResponses(in []model.DataStagingTask) []DataStagingTaskResponse {
- out := make([]DataStagingTaskResponse, 0, len(in))
- for i := range in {
- out = append(out, ToDataStagingTaskResponse(&in[i]))
- }
- return out
-}
-
-// JobSubmissionTaskRequest is the create/update payload for a submission task.
-//
-// jobId is writable rather than server-generated: it is the scheduler's identifier for
-// the submitted job, learned at submission time and recorded here afterwards.
-type JobSubmissionTaskRequest struct {
- JobID *string `json:"jobId"`
- OnFailure *model.OnFailureAction `json:"onFailure"`
- RetryCount *int `json:"retryCount"`
- TaskOrder *int `json:"taskOrder"`
-}
-
-// Validate implements httpx.Validator.
-func (r *JobSubmissionTaskRequest) Validate() []httpx.FieldError {
- var c httpx.Constraints
- taskCommon(&c, r.OnFailure, r.RetryCount, r.TaskOrder)
- return c.Fields()
-}
-
-func ApplyJobSubmissionTaskRequest(dst *model.JobSubmissionTask, src *JobSubmissionTaskRequest) {
- dst.JobId = src.JobID
- dst.OnFailure = src.OnFailure
- dst.RetryCount = src.RetryCount
- dst.TaskOrder = src.TaskOrder
-}
-
-// JobSubmissionTaskResponse is the read model for a submission task.
-type JobSubmissionTaskResponse struct {
- TaskID string `json:"taskId"`
- ProcessID *string `json:"processId"`
- ProcessType *string `json:"processType"`
-
- JobID *string `json:"jobId"`
- OnFailure *model.OnFailureAction `json:"onFailure"`
- RetryCount *int `json:"retryCount"`
- TaskOrder *int `json:"taskOrder"`
-}
-
-func ToJobSubmissionTaskResponse(t *model.JobSubmissionTask) JobSubmissionTaskResponse {
- return JobSubmissionTaskResponse{
- TaskID: t.ID,
- ProcessID: t.ProcessID,
- ProcessType: t.ProcessType,
- JobID: t.JobId,
- OnFailure: t.OnFailure,
- RetryCount: t.RetryCount,
- TaskOrder: t.TaskOrder,
- }
-}
-
-func ToJobSubmissionTaskResponses(in []model.JobSubmissionTask) []JobSubmissionTaskResponse {
- out := make([]JobSubmissionTaskResponse, 0, len(in))
- for i := range in {
- out = append(out, ToJobSubmissionTaskResponse(&in[i]))
- }
- return out
-}
-
-// JobMonitoringTaskRequest is the create/update payload for a monitoring task.
-type JobMonitoringTaskRequest struct {
- JobID *string `json:"jobId"`
- OnFailure *model.OnFailureAction `json:"onFailure"`
- RetryCount *int `json:"retryCount"`
- TaskOrder *int `json:"taskOrder"`
-}
-
-// Validate implements httpx.Validator.
-func (r *JobMonitoringTaskRequest) Validate() []httpx.FieldError {
- var c httpx.Constraints
- taskCommon(&c, r.OnFailure, r.RetryCount, r.TaskOrder)
- return c.Fields()
-}
-
-func ApplyJobMonitoringTaskRequest(dst *model.JobMonitoringTask, src *JobMonitoringTaskRequest) {
- dst.JobId = src.JobID
- dst.OnFailure = src.OnFailure
- dst.RetryCount = src.RetryCount
- dst.TaskOrder = src.TaskOrder
-}
-
-// JobMonitoringTaskResponse is the read model for a monitoring task.
-type JobMonitoringTaskResponse struct {
- TaskID string `json:"taskId"`
- ProcessID *string `json:"processId"`
- ProcessType *string `json:"processType"`
-
- JobID *string `json:"jobId"`
- OnFailure *model.OnFailureAction `json:"onFailure"`
- RetryCount *int `json:"retryCount"`
- TaskOrder *int `json:"taskOrder"`
-}
-
-func ToJobMonitoringTaskResponse(t *model.JobMonitoringTask) JobMonitoringTaskResponse {
- return JobMonitoringTaskResponse{
- TaskID: t.ID,
- ProcessID: t.ProcessID,
- ProcessType: t.ProcessType,
- JobID: t.JobId,
- OnFailure: t.OnFailure,
- RetryCount: t.RetryCount,
- TaskOrder: t.TaskOrder,
- }
-}
-
-func ToJobMonitoringTaskResponses(in []model.JobMonitoringTask) []JobMonitoringTaskResponse {
- out := make([]JobMonitoringTaskResponse, 0, len(in))
- for i := range in {
- out = append(out, ToJobMonitoringTaskResponse(&in[i]))
- }
- return out
-}
-
-// InteractiveCommandTaskRequest is the create/update payload for a command task.
-//
-// output is writable for the same reason jobId is: it is the result of running the
-// command, recorded once whatever executed it knows the answer.
-type InteractiveCommandTaskRequest struct {
- Command *string `json:"command"`
- Output *string `json:"output"`
- OnFailure *model.OnFailureAction `json:"onFailure"`
- RetryCount *int `json:"retryCount"`
- TaskOrder *int `json:"taskOrder"`
-}
-
-// Validate implements httpx.Validator.
-func (r *InteractiveCommandTaskRequest) Validate() []httpx.FieldError {
- var c httpx.Constraints
- c.NotBlankPtr("command", "Command cannot be blank", r.Command)
- taskCommon(&c, r.OnFailure, r.RetryCount, r.TaskOrder)
- return c.Fields()
-}
-
-func ApplyInteractiveCommandTaskRequest(dst *model.InteractiveCommandTask, src *InteractiveCommandTaskRequest) {
- dst.Command = src.Command
- dst.Output = src.Output
- dst.OnFailure = src.OnFailure
- dst.RetryCount = src.RetryCount
- dst.TaskOrder = src.TaskOrder
-}
-
-// InteractiveCommandTaskResponse is the read model for a command task.
-type InteractiveCommandTaskResponse struct {
- TaskID string `json:"taskId"`
- ProcessID *string `json:"processId"`
- ProcessType *string `json:"processType"`
-
- Command *string `json:"command"`
- Output *string `json:"output"`
- OnFailure *model.OnFailureAction `json:"onFailure"`
- RetryCount *int `json:"retryCount"`
- TaskOrder *int `json:"taskOrder"`
-}
-
-func ToInteractiveCommandTaskResponse(t *model.InteractiveCommandTask) InteractiveCommandTaskResponse {
- return InteractiveCommandTaskResponse{
- TaskID: t.ID,
- ProcessID: t.ProcessID,
- ProcessType: t.ProcessType,
- Command: t.Command,
- Output: t.Output,
- OnFailure: t.OnFailure,
- RetryCount: t.RetryCount,
- TaskOrder: t.TaskOrder,
- }
-}
-
-func ToInteractiveCommandTaskResponses(in []model.InteractiveCommandTask) []InteractiveCommandTaskResponse {
- out := make([]InteractiveCommandTaskResponse, 0, len(in))
- for i := range in {
- out = append(out, ToInteractiveCommandTaskResponse(&in[i]))
- }
- return out
-}
diff --git a/api/process/model/batch_job_process.go b/api/process/model/batch.go
similarity index 65%
rename from api/process/model/batch_job_process.go
rename to api/process/model/batch.go
index e082cc1fab..3f902a1a79 100644
--- a/api/process/model/batch_job_process.go
+++ b/api/process/model/batch.go
@@ -2,28 +2,28 @@
package model
import (
+ applicationmodel "github.com/apache/airavata/api/application/model"
"github.com/google/uuid"
"gorm.io/gorm"
-
- applicationmodel "github.com/apache/airavata/api/application/model"
- iammodel "github.com/apache/airavata/api/iam/model"
)
// BatchJobProcess is one run of a BatchDeployment.
type BatchJobProcess struct {
- ID string `gorm:"column:process_id;primaryKey;type:varchar(36)" json:"processId"`
+ ID string `gorm:"column:batch_process_id;primaryKey;type:varchar(36)" json:"batchProcessId"`
+
+ ProcessID *string `gorm:"column:parent_process_id;type:varchar(36);index" json:"parentProcessId,omitempty"`
+ Process *Process `gorm:"references:ID;constraint:OnDelete:CASCADE,OnUpdate:CASCADE" json:"parentProcess,omitempty"`
DeploymentID *string `gorm:"column:deployment_id;type:varchar(36);index" json:"deploymentId,omitempty"`
Deployment *applicationmodel.BatchDeployment `gorm:"references:ID;constraint:OnDelete:RESTRICT,OnUpdate:CASCADE" json:"-"`
- OwnerID *string `gorm:"column:user_id;type:varchar(255);index" json:"userId,omitempty"`
- Owner *iammodel.User `gorm:"references:ID;constraint:OnDelete:RESTRICT,OnUpdate:CASCADE" json:"-"`
-
// Owned one-to-one, as on BatchDeployment: unique foreign key on this side, with
// the orphan removed by AfterDelete since the database cannot cascade outward.
BatchJobConfigID string `gorm:"column:batch_job_config_id;type:varchar(36);not null;uniqueIndex" json:"batchJobConfigId"`
BatchJobConfig *applicationmodel.BatchJobConfig `gorm:"references:ID;constraint:OnDelete:RESTRICT,OnUpdate:CASCADE" json:"batchJobConfig,omitempty"`
+ JobID *string `gorm:"column:job_id;type:varchar(255)" json:"jobId,omitempty"`
+ JobName *string `gorm:"column:job_name;type:varchar(255)" json:"jobName,omitempty"`
// The last status recorded for this process, if any.
//
// It carries no foreign key, deliberately. A status already points at its process,
@@ -35,7 +35,7 @@ type BatchJobProcess struct {
}
// TableName returns the table backing BatchJobProcess.
-func (BatchJobProcess) TableName() string { return "batch_job_processes" }
+func (BatchJobProcess) TableName() string { return "batch_processes" }
// BeforeCreate assigns a UUID when none was supplied.
func (p *BatchJobProcess) BeforeCreate(*gorm.DB) error {
@@ -45,16 +45,6 @@ func (p *BatchJobProcess) BeforeCreate(*gorm.DB) error {
return nil
}
-// BeforeDelete removes every recorded status ahead of the process itself.
-//
-// The statuses have to go first: their own ProcessID foreign key is RESTRICT, so the
-// database would refuse to delete a process any status still points at. LastStatusID
-// needs no clearing — it holds no constraint — and the row carrying it is about to be
-// deleted in this same transaction.
-func (p *BatchJobProcess) BeforeDelete(tx *gorm.DB) error {
- return tx.Where("process_id = ?", p.ID).Delete(&BatchJobProcessStatus{}).Error
-}
-
// AfterDelete removes the owned BatchJobConfig, standing in for JPA's orphanRemoval.
func (p *BatchJobProcess) AfterDelete(tx *gorm.DB) error {
if p.BatchJobConfigID == "" {
@@ -63,8 +53,3 @@ func (p *BatchJobProcess) AfterDelete(tx *gorm.DB) error {
return tx.Where("batch_job_config_id = ?", p.BatchJobConfigID).
Delete(&applicationmodel.BatchJobConfig{}).Error
}
-
-// OwnedBy reports whether userID launched this process.
-func (p *BatchJobProcess) OwnedBy(userID string) bool {
- return p.OwnerID != nil && *p.OwnerID == userID
-}
diff --git a/api/process/model/batch_job_process_status.go b/api/process/model/batch_job_process_status.go
deleted file mode 100644
index 94024b209d..0000000000
--- a/api/process/model/batch_job_process_status.go
+++ /dev/null
@@ -1,48 +0,0 @@
-package model
-
-import (
- "github.com/google/uuid"
- "gorm.io/gorm"
-)
-
-type BatchProcessStatusType string
-
-const (
- BatchProcessStatusTypeCreated BatchProcessStatusType = "CREATED"
- BatchProcessStatusTypeSubmitted BatchProcessStatusType = "SUBMITTED"
- BatchProcessStatusTypeRunning BatchProcessStatusType = "RUNNING"
- BatchProcessStatusTypeCompleted BatchProcessStatusType = "COMPLETED"
- BatchProcessStatusTypeFailed BatchProcessStatusType = "FAILED"
-)
-
-type BatchJobProcessStatus struct {
- ID string `gorm:"column:process_status_id;primaryKey;type:varchar(36)" json:"processStatusId"`
-
- ProcessID *string `gorm:"column:process_id;type:varchar(36);index" json:"processId,omitempty"`
- Process *BatchJobProcess `gorm:"references:ID;constraint:OnDelete:RESTRICT,OnUpdate:CASCADE" json:"-"`
-
- Status *BatchProcessStatusType `gorm:"column:status;type:varchar(255)" json:"status,omitempty"`
- Log *string `gorm:"column:log;type:text" json:"log,omitempty"`
-
- // The time the status was recorded, in milliseconds since the epoch.
- Timestamp *int64 `gorm:"column:timestamp" json:"timestamp,omitempty"`
-}
-
-// TableName returns the table backing BatchJobProcessStatus.
-func (BatchJobProcessStatus) TableName() string { return "batch_job_process_statuses" }
-
-func (s *BatchJobProcessStatus) BeforeCreate(*gorm.DB) error {
- if s.ID == "" {
- s.ID = uuid.NewString()
- }
- return nil
-}
-
-func (s *BatchJobProcessStatus) AfterCreate(tx *gorm.DB) error {
- if s.ProcessID == nil || *s.ProcessID == "" {
- return nil
- }
- return tx.Model(&BatchJobProcess{}).
- Where("process_id = ?", *s.ProcessID).
- Update("last_status_id", s.ID).Error
-}
diff --git a/api/process/model/process.go b/api/process/model/process.go
new file mode 100644
index 0000000000..7c267afd7a
--- /dev/null
+++ b/api/process/model/process.go
@@ -0,0 +1,121 @@
+package model
+
+import (
+ application "github.com/apache/airavata/api/application/model"
+ iammodel "github.com/apache/airavata/api/iam/model"
+ "github.com/google/uuid"
+ "gorm.io/gorm"
+)
+
+type TemplateInputMapping struct {
+ TemplateInputMappingID string `gorm:"column:template_input_mapping_id;primaryKey;type:varchar(36)" json:"templateInputMappingId"`
+
+ // The template this mapping belongs to.
+ TemplateInputID *string `gorm:"column:template_input_id;type:varchar(36);index" json:"templateInputId,omitempty"`
+ TemplateInput *application.TemplateInput `gorm:"references:TemplateInputID;constraint:OnDelete:CASCADE,OnUpdate:CASCADE" json:"templateInput,omitempty"`
+
+ ProcessID *string `gorm:"column:process_id;type:varchar(36);index" json:"processId,omitempty"`
+ Process *Process `gorm:"references:ID;constraint:OnDelete:CASCADE,OnUpdate:CASCADE" json:"process,omitempty"`
+
+ // Value is a JSON document, either {"value": "..."} for a single value or
+ // {"values": [...]} for a list.
+ Value *string `gorm:"column:value;type:text" json:"value,omitempty"`
+}
+
+type TemplateOutputMapping struct {
+ TemplateOutputMappingID string `gorm:"column:template_output_mapping_id;primaryKey;type:varchar(36)" json:"templateOutputMappingId"`
+
+ // The template this mapping belongs to.
+ TemplateOutputID *string `gorm:"column:template_output_id;type:varchar(36);index" json:"templateOutputId,omitempty"`
+ TemplateOutput *application.TemplateOutput `gorm:"references:TemplateOutputID;constraint:OnDelete:CASCADE,OnUpdate:CASCADE" json:"templateOutput,omitempty"`
+
+ ProcessID *string `gorm:"column:process_id;type:varchar(36);index" json:"processId,omitempty"`
+ Process *Process `gorm:"references:ID;constraint:OnDelete:CASCADE,OnUpdate:CASCADE" json:"process,omitempty"`
+ // Value is a JSON document, either {"value": "..."} for a single value or
+ // {"values": [...]} for a list.
+ Value *string `gorm:"column:value;type:text" json:"value,omitempty"`
+}
+
+type ProcessType string
+
+const (
+ ProcessTypeBatchJob ProcessType = "BATCH_JOB"
+ ProcessTypeTask ProcessType = "CLOUD_JOB"
+)
+
+type Process struct {
+ ID string `gorm:"column:process_id;primaryKey;type:varchar(36)" json:"processId"`
+
+ OwnerID *string `gorm:"column:user_id;type:varchar(255);index" json:"userId,omitempty"`
+ Owner *iammodel.User `gorm:"references:ID;constraint:OnDelete:RESTRICT,OnUpdate:CASCADE" json:"-"`
+
+ // The process type, e.g. "BATCH_JOB". This is a plain string rather than an enum
+ // because the set of process kinds is expected to grow.
+ ProcessType *ProcessType `gorm:"column:process_type;type:varchar(32)" json:"processType,omitempty"`
+ InputMappings []*TemplateInputMapping `gorm:"foreignKey:ProcessID" json:"inputMappings,omitempty"`
+ OutputMappings []*TemplateOutputMapping `gorm:"foreignKey:ProcessID" json:"outputMappings,omitempty"`
+}
+
+// TableName returns the table backing TemplateInputMapping.
+func (TemplateInputMapping) TableName() string { return "process_template_input_mappings" }
+
+// TableName returns the table backing TemplateOutputMapping.
+func (TemplateOutputMapping) TableName() string { return "process_template_output_mappings" }
+
+// BeforeCreate assigns a UUID when none was supplied.
+func (m *TemplateInputMapping) BeforeCreate(*gorm.DB) error {
+ if m.TemplateInputMappingID == "" {
+ m.TemplateInputMappingID = uuid.NewString()
+ }
+ return nil
+}
+
+// BeforeCreate assigns a UUID when none was supplied.
+func (m *TemplateOutputMapping) BeforeCreate(*gorm.DB) error {
+ if m.TemplateOutputMappingID == "" {
+ m.TemplateOutputMappingID = uuid.NewString()
+ }
+ return nil
+}
+
+type ProcessStatusType string
+
+const (
+ ProcessStatusTypeCreated ProcessStatusType = "CREATED"
+ ProcessStatusTypeSubmitted ProcessStatusType = "SUBMITTED"
+ ProcessStatusTypeRunning ProcessStatusType = "RUNNING"
+ ProcessStatusTypeCompleted ProcessStatusType = "COMPLETED"
+ ProcessStatusTypeFailed ProcessStatusType = "FAILED"
+)
+
+type ProcessStatus struct {
+ ID string `gorm:"column:process_status_id;primaryKey;type:varchar(36)" json:"processStatusId"`
+
+ ProcessID *string `gorm:"column:process_id;type:varchar(36);index" json:"processId,omitempty"`
+ Process *Process `gorm:"references:ID;constraint:OnDelete:RESTRICT,OnUpdate:CASCADE" json:"-"`
+
+ Status *ProcessStatusType `gorm:"column:status;type:varchar(255)" json:"status,omitempty"`
+ Log *string `gorm:"column:log;type:text" json:"log,omitempty"`
+
+ // The time the status was recorded, in milliseconds since the epoch.
+ Timestamp *int64 `gorm:"column:timestamp" json:"timestamp,omitempty"`
+}
+
+// TableName returns the table backing ProcessStatus.
+func (ProcessStatus) TableName() string { return "process_statuses" }
+
+func (s *ProcessStatus) BeforeCreate(*gorm.DB) error {
+ if s.ID == "" {
+ s.ID = uuid.NewString()
+ }
+ return nil
+}
+
+func (s *ProcessStatus) AfterCreate(tx *gorm.DB) error {
+ if s.ProcessID == nil || *s.ProcessID == "" {
+ return nil
+ }
+ return tx.Model(&Process{}).
+ Where("process_id = ?", *s.ProcessID).
+ Update("last_status_id", s.ID).Error
+}
\ No newline at end of file
diff --git a/api/process/model/task.go b/api/process/model/task.go
index 89a4f1e670..ed898e96dc 100644
--- a/api/process/model/task.go
+++ b/api/process/model/task.go
@@ -5,11 +5,6 @@ import (
"gorm.io/gorm"
)
-// ProcessTypeBatchJob is the ProcessType a task carries when it belongs to a
-// BatchJobProcess. The column is a plain string rather than an enum because the set of
-// process kinds is expected to grow; this is the only member so far.
-const ProcessTypeBatchJob = "BATCH_JOB"
-
type OnFailureAction string
const (
@@ -46,9 +41,8 @@ func (t DataStorageType) Valid() bool {
type DataStagingTask struct {
ID string `gorm:"column:task_id;primaryKey;type:varchar(36)" json:"taskId"`
- // The process this task belongs to.
- ProcessID *string `gorm:"column:process_id;type:varchar(36);index" json:"processId,omitempty"`
- ProcessType *string `gorm:"column:process_type;type:varchar(32)" json:"processType,omitempty"`
+ ProcessID *string `gorm:"column:process_id;type:varchar(36);index" json:"processId,omitempty"`
+ Process *Process `gorm:"references:ID;constraint:OnDelete:CASCADE,OnUpdate:CASCADE" json:"process,omitempty"`
SourceDataStorageID *string `gorm:"column:source_data_storage_id;type:varchar(36);index" json:"sourceDataStorageId,omitempty"`
SourceCredentialID *string `gorm:"column:source_credential_id;type:varchar(36);index" json:"sourceCredentialId,omitempty"`
@@ -75,9 +69,8 @@ type DataStagingTask struct {
type JobSubmissionTask struct {
ID string `gorm:"column:task_id;primaryKey;type:varchar(36)" json:"taskId"`
- // The process this task belongs to.
- ProcessID *string `gorm:"column:process_id;type:varchar(36);index" json:"processId,omitempty"`
- ProcessType *string `gorm:"column:process_type;type:varchar(32)" json:"processType,omitempty"`
+ ProcessID *string `gorm:"column:process_id;type:varchar(36);index" json:"processId,omitempty"`
+ Process *Process `gorm:"references:ID;constraint:OnDelete:CASCADE,OnUpdate:CASCADE" json:"process,omitempty"`
JobId *string `gorm:"column:job_id;type:varchar(255)" json:"jobId,omitempty"`
OnFailure *OnFailureAction `gorm:"column:on_failure;type:varchar(32)" json:"onFailure,omitempty"`
@@ -89,9 +82,8 @@ type JobSubmissionTask struct {
type JobMonitoringTask struct {
ID string `gorm:"column:task_id;primaryKey;type:varchar(36)" json:"taskId"`
- // The process this task belongs to.
- ProcessID *string `gorm:"column:process_id;type:varchar(36);index" json:"processId,omitempty"`
- ProcessType *string `gorm:"column:process_type;type:varchar(32)" json:"processType,omitempty"`
+ ProcessID *string `gorm:"column:process_id;type:varchar(36);index" json:"processId,omitempty"`
+ Process *Process `gorm:"references:ID;constraint:OnDelete:CASCADE,OnUpdate:CASCADE" json:"process,omitempty"`
JobId *string `gorm:"column:job_id;type:varchar(255)" json:"jobId,omitempty"`
OnFailure *OnFailureAction `gorm:"column:on_failure;type:varchar(32)" json:"onFailure,omitempty"`
@@ -104,9 +96,8 @@ type JobMonitoringTask struct {
type InteractiveCommandTask struct {
ID string `gorm:"column:task_id;primaryKey;type:varchar(36)" json:"taskId"`
- // The process this task belongs to.
- ProcessID *string `gorm:"column:process_id;type:varchar(36);index" json:"processId,omitempty"`
- ProcessType *string `gorm:"column:process_type;type:varchar(32)" json:"processType,omitempty"`
+ ProcessID *string `gorm:"column:process_id;type:varchar(36);index" json:"processId,omitempty"`
+ Process *Process `gorm:"references:ID;constraint:OnDelete:CASCADE,OnUpdate:CASCADE" json:"process,omitempty"`
Command *string `gorm:"column:command;type:text" json:"command,omitempty"`
Output *string `gorm:"column:output;type:text" json:"output,omitempty"`
diff --git a/api/process/repository/process.go b/api/process/repository/process.go
deleted file mode 100644
index dee3300b51..0000000000
--- a/api/process/repository/process.go
+++ /dev/null
@@ -1,72 +0,0 @@
-// Package repository reads and writes batch job processes and their status history.
-package repository
-
-import (
- "context"
-
- "gorm.io/gorm"
-
- applicationmodel "github.com/apache/airavata/api/application/model"
- model "github.com/apache/airavata/api/process/model"
-)
-
-// ProcessRepository reads and writes batch job processes.
-type ProcessRepository struct{ db *gorm.DB }
-
-// NewProcessRepository returns a repository backed by db.
-func NewProcessRepository(db *gorm.DB) *ProcessRepository { return &ProcessRepository{db: db} }
-
-// WithTx returns a repository bound to tx.
-func (r *ProcessRepository) WithTx(tx *gorm.DB) *ProcessRepository { return &ProcessRepository{db: tx} }
-
-// FindAll returns every process with its owned resource request.
-func (r *ProcessRepository) FindAll(ctx context.Context) ([]model.BatchJobProcess, error) {
- var out []model.BatchJobProcess
- err := r.db.WithContext(ctx).Preload("BatchJobConfig").Find(&out).Error
- return out, err
-}
-
-// FindByDeploymentID returns every process of one deployment.
-func (r *ProcessRepository) FindByDeploymentID(ctx context.Context, deploymentID string) ([]model.BatchJobProcess, error) {
- var out []model.BatchJobProcess
- err := r.db.WithContext(ctx).Preload("BatchJobConfig").
- Where("deployment_id = ?", deploymentID).Find(&out).Error
- return out, err
-}
-
-// FindByOwnerID returns every process launched by one user.
-//
-// Declared in the Java repository but never called — there is no "my processes"
-// endpoint, which is why process reads are not owner-scoped at all.
-func (r *ProcessRepository) FindByOwnerID(ctx context.Context, userID string) ([]model.BatchJobProcess, error) {
- var out []model.BatchJobProcess
- err := r.db.WithContext(ctx).Preload("BatchJobConfig").
- Where("user_id = ?", userID).Find(&out).Error
- return out, err
-}
-
-// FindByID returns one process, or gorm.ErrRecordNotFound.
-func (r *ProcessRepository) FindByID(ctx context.Context, id string) (*model.BatchJobProcess, error) {
- var out model.BatchJobProcess
- err := r.db.WithContext(ctx).Preload("BatchJobConfig").
- First(&out, "process_id = ?", id).Error
- if err != nil {
- return nil, err
- }
- return &out, nil
-}
-
-// Save writes the process row, leaving its owned config to SaveConfig.
-func (r *ProcessRepository) Save(ctx context.Context, p *model.BatchJobProcess) error {
- return r.db.WithContext(ctx).Omit("BatchJobConfig").Save(p).Error
-}
-
-// SaveConfig inserts or updates the owned resource request.
-func (r *ProcessRepository) SaveConfig(ctx context.Context, c *applicationmodel.BatchJobConfig) error {
- return r.db.WithContext(ctx).Save(c).Error
-}
-
-// Delete removes a process; the AfterDelete hook removes its owned config.
-func (r *ProcessRepository) Delete(ctx context.Context, p *model.BatchJobProcess) error {
- return r.db.WithContext(ctx).Delete(p).Error
-}
diff --git a/api/process/repository/status.go b/api/process/repository/status.go
deleted file mode 100644
index 22dfe2ae87..0000000000
--- a/api/process/repository/status.go
+++ /dev/null
@@ -1,48 +0,0 @@
-package repository
-
-import (
- "context"
-
- "gorm.io/gorm"
-
- model "github.com/apache/airavata/api/process/model"
-)
-
-// StatusRepository reads and writes batch job process statuses.
-//
-// There is no Update: statuses are append-only history, never corrected in place.
-type StatusRepository struct{ db *gorm.DB }
-
-// NewStatusRepository returns a repository backed by db.
-func NewStatusRepository(db *gorm.DB) *StatusRepository { return &StatusRepository{db: db} }
-
-// WithTx returns a repository bound to tx.
-func (r *StatusRepository) WithTx(tx *gorm.DB) *StatusRepository { return &StatusRepository{db: tx} }
-
-// FindByProcessID returns every status recorded for one process, oldest first.
-func (r *StatusRepository) FindByProcessID(ctx context.Context, processID string) ([]model.BatchJobProcessStatus, error) {
- var out []model.BatchJobProcessStatus
- err := r.db.WithContext(ctx).
- Where("process_id = ?", processID).
- Order("timestamp").
- Find(&out).Error
- return out, err
-}
-
-// FindByIDAndProcessID returns one status scoped to its process, or
-// gorm.ErrRecordNotFound if the id does not belong to that process.
-func (r *StatusRepository) FindByIDAndProcessID(ctx context.Context, id, processID string) (*model.BatchJobProcessStatus, error) {
- var out model.BatchJobProcessStatus
- err := r.db.WithContext(ctx).
- First(&out, "process_status_id = ? AND process_id = ?", id, processID).Error
- if err != nil {
- return nil, err
- }
- return &out, nil
-}
-
-// Create inserts a status row. BeforeCreate assigns its id and AfterCreate repoints
-// the owning process's LastStatusID at it.
-func (r *StatusRepository) Create(ctx context.Context, s *model.BatchJobProcessStatus) error {
- return r.db.WithContext(ctx).Create(s).Error
-}
diff --git a/api/process/repository/task.go b/api/process/repository/task.go
deleted file mode 100644
index 2c6577db30..0000000000
--- a/api/process/repository/task.go
+++ /dev/null
@@ -1,94 +0,0 @@
-package repository
-
-import (
- "context"
-
- "gorm.io/gorm"
-
- model "github.com/apache/airavata/api/process/model"
-)
-
-// TaskRepository reads and writes one kind of process task.
-//
-// The four kinds differ in payload but not in identity: every one is keyed by task_id,
-// scoped by process_id and ordered by task_order. That is the whole of the table
-// access, so it is written once here and instantiated per kind rather than copied four
-// times. GORM resolves the table from T's TableName.
-type TaskRepository[T any] struct{ db *gorm.DB }
-
-// NewTaskRepository returns a repository over the table backing T.
-func NewTaskRepository[T any](db *gorm.DB) *TaskRepository[T] {
- return &TaskRepository[T]{db: db}
-}
-
-// WithTx returns a repository bound to tx.
-func (r *TaskRepository[T]) WithTx(tx *gorm.DB) *TaskRepository[T] {
- return &TaskRepository[T]{db: tx}
-}
-
-// FindByProcessID returns every task of one process in execution order.
-//
-// task_order is nullable, so rows without one sort last rather than jumping ahead of
-// an explicitly ordered step; task_id breaks ties so the order is stable between
-// calls.
-func (r *TaskRepository[T]) FindByProcessID(ctx context.Context, processID string) ([]T, error) {
- var out []T
- err := r.db.WithContext(ctx).
- Where("process_id = ?", processID).
- Order("task_order IS NULL, task_order, task_id").
- Find(&out).Error
- return out, err
-}
-
-// FindByIDAndProcessID returns one task scoped to its process, or
-// gorm.ErrRecordNotFound if the id does not belong to that process.
-func (r *TaskRepository[T]) FindByIDAndProcessID(ctx context.Context, id, processID string) (*T, error) {
- var out T
- err := r.db.WithContext(ctx).
- Where("task_id = ? AND process_id = ?", id, processID).
- First(&out).Error
- if err != nil {
- return nil, err
- }
- return &out, nil
-}
-
-// Save inserts or updates a task.
-func (r *TaskRepository[T]) Save(ctx context.Context, t *T) error {
- return r.db.WithContext(ctx).Save(t).Error
-}
-
-// Delete removes a task.
-func (r *TaskRepository[T]) Delete(ctx context.Context, t *T) error {
- return r.db.WithContext(ctx).Delete(t).Error
-}
-
-// The four instantiations. They exist so the wiring reads in entity terms rather than
-// in type parameters, and so a future kind-specific query has somewhere to live.
-type (
- DataStagingTaskRepository = TaskRepository[model.DataStagingTask]
- JobSubmissionTaskRepository = TaskRepository[model.JobSubmissionTask]
- JobMonitoringTaskRepository = TaskRepository[model.JobMonitoringTask]
- InteractiveCommandTaskRepository = TaskRepository[model.InteractiveCommandTask]
-)
-
-// NewDataStagingTaskRepository returns a repository over data_staging_tasks.
-func NewDataStagingTaskRepository(db *gorm.DB) *DataStagingTaskRepository {
- return NewTaskRepository[model.DataStagingTask](db)
-}
-
-// NewJobSubmissionTaskRepository returns a repository over job_submission_tasks.
-func NewJobSubmissionTaskRepository(db *gorm.DB) *JobSubmissionTaskRepository {
- return NewTaskRepository[model.JobSubmissionTask](db)
-}
-
-// NewJobMonitoringTaskRepository returns a repository over job_monitoring_tasks.
-func NewJobMonitoringTaskRepository(db *gorm.DB) *JobMonitoringTaskRepository {
- return NewTaskRepository[model.JobMonitoringTask](db)
-}
-
-// NewInteractiveCommandTaskRepository returns a repository over
-// interactive_command_tasks.
-func NewInteractiveCommandTaskRepository(db *gorm.DB) *InteractiveCommandTaskRepository {
- return NewTaskRepository[model.InteractiveCommandTask](db)
-}
diff --git a/api/process/service/process.go b/api/process/service/process.go
deleted file mode 100644
index 9771f53662..0000000000
--- a/api/process/service/process.go
+++ /dev/null
@@ -1,197 +0,0 @@
-// Package service holds the process vertical's business rules: submitting a job under
-// the caller's own identity, and recording what happened to it.
-package service
-
-import (
- "context"
- "errors"
-
- "gorm.io/gorm"
-
- "github.com/apache/airavata/internal/auth"
- "github.com/apache/airavata/internal/httpx"
-
- applicationdto "github.com/apache/airavata/api/application/dto"
- applicationmodel "github.com/apache/airavata/api/application/model"
- applicationrepo "github.com/apache/airavata/api/application/repository"
- iamrepo "github.com/apache/airavata/api/iam/repository"
- dto "github.com/apache/airavata/api/process/dto"
- model "github.com/apache/airavata/api/process/model"
- "github.com/apache/airavata/api/process/repository"
-)
-
-func notFoundAs(err error, format string, args ...any) error {
- if errors.Is(err, gorm.ErrRecordNotFound) {
- return httpx.NotFound(format, args...)
- }
- return err
-}
-
-// ProcessService manages batch job processes: individual runs of a deployment.
-//
-// Note the asymmetry with SSH endpoint credentials and SCP data. Submitting is
-// self-service and ownership is taken from the token, but reads are not owner-scoped:
-// listing by deployment and fetching by id carry no authorisation at all, matching
-// the Java service. That is worth revisiting, but it is the behaviour clients have.
-type ProcessService struct {
- db *gorm.DB
- processes *repository.ProcessRepository
- deployments *applicationrepo.BatchDeploymentRepository
- users *iamrepo.UserRepository
- statuses *StatusService
-}
-
-// NewProcessService returns a batch job process service.
-func NewProcessService(
- db *gorm.DB,
- processes *repository.ProcessRepository,
- deployments *applicationrepo.BatchDeploymentRepository,
- users *iamrepo.UserRepository,
- statuses *StatusService,
-) *ProcessService {
- return &ProcessService{db: db, processes: processes, deployments: deployments, users: users, statuses: statuses}
-}
-
-// List returns every process across every user. Admin only.
-func (s *ProcessService) List(ctx context.Context) ([]dto.Response, error) {
- if _, err := auth.RequireAdmin(ctx); err != nil {
- return nil, err
- }
- found, err := s.processes.FindAll(ctx)
- if err != nil {
- return nil, err
- }
- return dto.ToResponses(found), nil
-}
-
-// ListByDeployment returns every process of one deployment.
-func (s *ProcessService) ListByDeployment(ctx context.Context, deploymentID string) ([]dto.Response, error) {
- found, err := s.processes.FindByDeploymentID(ctx, deploymentID)
- if err != nil {
- return nil, err
- }
- return dto.ToResponses(found), nil
-}
-
-// Get returns one process.
-func (s *ProcessService) Get(ctx context.Context, id string) (*dto.Response, error) {
- proc, err := s.processes.FindByID(ctx, id)
- if err != nil {
- return nil, notFoundAs(err, "Process not found: %s", id)
- }
- out := dto.ToResponse(proc)
- return &out, nil
-}
-
-// Create submits a process for the calling user.
-//
-// Any authenticated caller may submit for themselves — this is self-service, not an
-// administrative operation — and the owner always comes from the token.
-func (s *ProcessService) Create(ctx context.Context, req *dto.Request) (*dto.Response, error) {
- principal, err := auth.RequireAuthenticated(ctx)
- if err != nil {
- return nil, err
- }
-
- var out dto.Response
- err = s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
- processes, deployments, users := s.processes.WithTx(tx), s.deployments.WithTx(tx), s.users.WithTx(tx)
-
- deployment, err := deployments.FindByID(ctx, req.DeploymentID)
- if err != nil {
- return notFoundAs(err, "Deployment not found: %s", req.DeploymentID)
- }
- owner, err := users.FindByID(ctx, principal.Name)
- if err != nil {
- return notFoundAs(err, "No user record found for authenticated principal: %s", principal.Name)
- }
-
- config := &applicationmodel.BatchJobConfig{}
- applicationdto.ApplyBatchJobConfigRequest(config, req.BatchJobConfig)
- if err := processes.SaveConfig(ctx, config); err != nil {
- return err
- }
-
- proc := &model.BatchJobProcess{
- DeploymentID: &deployment.ID,
- OwnerID: &owner.ID,
- BatchJobConfigID: config.ID,
- BatchJobConfig: config,
- }
- if err := processes.Save(ctx, proc); err != nil {
- return err
- }
-
- // Recorded in the same transaction as the process itself, so a caller never
- // observes a process that exists but has no status yet.
- if _, err := s.statuses.RecordTx(ctx, tx, proc.ID, model.BatchProcessStatusTypeCreated, nil); err != nil {
- return err
- }
-
- out = dto.ToResponse(proc)
- return nil
- })
- if err != nil {
- return nil, err
- }
- return &out, nil
-}
-
-// Update is an administrative correction of a process's deployment or resources.
-//
-// Ownership is deliberately immutable: re-deriving it from the caller's token would
-// reassign the process to whichever admin issued the request.
-func (s *ProcessService) Update(ctx context.Context, id string, req *dto.Request) (*dto.Response, error) {
- if _, err := auth.RequireAdmin(ctx); err != nil {
- return nil, err
- }
-
- var out dto.Response
- err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
- processes, deployments := s.processes.WithTx(tx), s.deployments.WithTx(tx)
-
- proc, err := processes.FindByID(ctx, id)
- if err != nil {
- return notFoundAs(err, "Process not found: %s", id)
- }
- deployment, err := deployments.FindByID(ctx, req.DeploymentID)
- if err != nil {
- return notFoundAs(err, "Deployment not found: %s", req.DeploymentID)
- }
-
- config := proc.BatchJobConfig
- if config == nil {
- config = &applicationmodel.BatchJobConfig{ID: proc.BatchJobConfigID}
- }
- applicationdto.ApplyBatchJobConfigRequest(config, req.BatchJobConfig)
- if err := processes.SaveConfig(ctx, config); err != nil {
- return err
- }
-
- proc.DeploymentID = &deployment.ID
- proc.BatchJobConfigID = config.ID
- proc.BatchJobConfig = config
-
- if err := processes.Save(ctx, proc); err != nil {
- return err
- }
- out = dto.ToResponse(proc)
- return nil
- })
- if err != nil {
- return nil, err
- }
- return &out, nil
-}
-
-// Delete removes a process and its owned resource request. Admin only.
-func (s *ProcessService) Delete(ctx context.Context, id string) error {
- if _, err := auth.RequireAdmin(ctx); err != nil {
- return err
- }
- proc, err := s.processes.FindByID(ctx, id)
- if err != nil {
- return notFoundAs(err, "Process not found: %s", id)
- }
- return s.processes.Delete(ctx, proc)
-}
diff --git a/api/process/service/status.go b/api/process/service/status.go
deleted file mode 100644
index 5606365761..0000000000
--- a/api/process/service/status.go
+++ /dev/null
@@ -1,105 +0,0 @@
-package service
-
-import (
- "context"
- "time"
-
- "gorm.io/gorm"
-
- "github.com/apache/airavata/internal/ptr"
-
- dto "github.com/apache/airavata/api/process/dto"
- model "github.com/apache/airavata/api/process/model"
- "github.com/apache/airavata/api/process/repository"
-)
-
-// StatusService records and reads BatchJobProcess status history.
-//
-// There is no create/update HTTP endpoint for statuses, by design: they are recorded
-// internally rather than accepted from a client request body — an initial CREATED
-// status when a process is submitted (see ProcessService.Create), and later transitions by
-// whatever submits and monitors the actual job. Reads are still a service concern
-// other code can call directly, and (via StatusController) a read-only API a client
-// can poll for progress.
-type StatusService struct {
- db *gorm.DB
- statuses *repository.StatusRepository
- processes *repository.ProcessRepository
-}
-
-// NewStatusService returns a status service.
-func NewStatusService(db *gorm.DB, statuses *repository.StatusRepository, processes *repository.ProcessRepository) *StatusService {
- return &StatusService{db: db, statuses: statuses, processes: processes}
-}
-
-// Record appends a new status for a process, in its own transaction.
-//
-// This is the entry point for a caller that is not already inside a transaction of
-// its own — the normal case for a service in another package recording a transition
-// it observed (e.g. a job monitor noticing a submission succeeded). It does not check
-// authorisation: recording a status is an internal operation performed by trusted
-// service code, never taken directly from a client request.
-func (s *StatusService) Record(ctx context.Context, processID string, status model.BatchProcessStatusType, log *string) (*dto.StatusResponse, error) {
- var out dto.StatusResponse
- err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
- created, err := s.RecordTx(ctx, tx, processID, status, log)
- if err != nil {
- return err
- }
- out = dto.ToStatusResponse(created)
- return nil
- })
- if err != nil {
- return nil, err
- }
- return &out, nil
-}
-
-// RecordTx appends a new status inside an already-open transaction, so a caller that
-// is already inside one of its own — ProcessService.Create recording the initial CREATED
-// status alongside the process it belongs to, or another service's own multi-step
-// write — commits or rolls back the status together with the rest of its work.
-func (s *StatusService) RecordTx(ctx context.Context, tx *gorm.DB, processID string, status model.BatchProcessStatusType, log *string) (*model.BatchJobProcessStatus, error) {
- if _, err := s.processes.WithTx(tx).FindByID(ctx, processID); err != nil {
- return nil, notFoundAs(err, "Process not found: %s", processID)
- }
-
- created := &model.BatchJobProcessStatus{
- ProcessID: &processID,
- Status: &status,
- Log: log,
- Timestamp: ptr.To(time.Now().UnixMilli()),
- }
- if err := s.statuses.WithTx(tx).Create(ctx, created); err != nil {
- return nil, err
- }
- return created, nil
-}
-
-// ListForProcess returns every recorded status of a process, oldest first.
-//
-// Not owner-scoped, matching the rest of this package's reads (see ProcessService's doc
-// comment) — the Java service carried no authorisation on process status either.
-func (s *StatusService) ListForProcess(ctx context.Context, processID string) ([]dto.StatusResponse, error) {
- if _, err := s.processes.FindByID(ctx, processID); err != nil {
- return nil, notFoundAs(err, "Process not found: %s", processID)
- }
- found, err := s.statuses.FindByProcessID(ctx, processID)
- if err != nil {
- return nil, err
- }
- return dto.ToStatusResponses(found), nil
-}
-
-// Get returns one status, scoped to the process it must belong to.
-func (s *StatusService) Get(ctx context.Context, processID, statusID string) (*dto.StatusResponse, error) {
- if _, err := s.processes.FindByID(ctx, processID); err != nil {
- return nil, notFoundAs(err, "Process not found: %s", processID)
- }
- found, err := s.statuses.FindByIDAndProcessID(ctx, statusID, processID)
- if err != nil {
- return nil, notFoundAs(err, "Status not found: %s in process %s", statusID, processID)
- }
- out := dto.ToStatusResponse(found)
- return &out, nil
-}
diff --git a/api/process/service/task.go b/api/process/service/task.go
deleted file mode 100644
index 98ba5b9d8e..0000000000
--- a/api/process/service/task.go
+++ /dev/null
@@ -1,209 +0,0 @@
-package service
-
-import (
- "context"
-
- "gorm.io/gorm"
-
- "github.com/apache/airavata/internal/auth"
- "github.com/apache/airavata/internal/httpx"
-
- dto "github.com/apache/airavata/api/process/dto"
- model "github.com/apache/airavata/api/process/model"
- "github.com/apache/airavata/api/process/repository"
-)
-
-// TaskService manages one kind of task as a sub-resource of its process.
-//
-// The four kinds are the same service with a different payload — scope by process,
-// authorise against that process, order by task_order — so the rules are written once
-// and instantiated per kind. What differs is supplied at construction: how a request
-// is applied to a row, and what the kind is called in an error message.
-//
-// Unlike the rest of this package, reads here are owner-scoped. Process and status
-// reads are open for compatibility with clients the Java service already had; these
-// endpoints are new, and a task carries filesystem paths and shell commands, so there
-// is no reason to expose them more widely than the run they belong to.
-type TaskService[T any, Req any] struct {
- db *gorm.DB
- tasks *repository.TaskRepository[T]
- processes *repository.ProcessRepository
-
- // name is how the kind is named in a not-found message, e.g. "Data staging task".
- name string
- // apply copies a validated request onto a row.
- apply func(dst *T, src *Req)
- // stamp records which process a new row belongs to.
- stamp func(dst *T, processID string)
-}
-
-// ListForProcess returns every task of one process, in execution order.
-func (s *TaskService[T, Req]) ListForProcess(ctx context.Context, processID string) ([]T, error) {
- if _, err := s.requireProcess(ctx, processID); err != nil {
- return nil, err
- }
- return s.tasks.FindByProcessID(ctx, processID)
-}
-
-// Get returns one task of one process.
-func (s *TaskService[T, Req]) Get(ctx context.Context, processID, taskID string) (*T, error) {
- if _, err := s.requireProcess(ctx, processID); err != nil {
- return nil, err
- }
- return s.requireTask(ctx, s.tasks, processID, taskID)
-}
-
-// Create adds a task to a process.
-func (s *TaskService[T, Req]) Create(ctx context.Context, processID string, req *Req) (*T, error) {
- var out *T
- err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
- if _, err := s.requireProcessTx(ctx, tx, processID); err != nil {
- return err
- }
-
- var task T
- s.stamp(&task, processID)
- s.apply(&task, req)
- if err := s.tasks.WithTx(tx).Save(ctx, &task); err != nil {
- return err
- }
- out = &task
- return nil
- })
- if err != nil {
- return nil, err
- }
- return out, nil
-}
-
-// Update changes a task of a process.
-//
-// The process it belongs to is left alone: both ids come from the path, so a task can
-// be edited but never moved to a different run.
-func (s *TaskService[T, Req]) Update(ctx context.Context, processID, taskID string, req *Req) (*T, error) {
- var out *T
- err := s.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
- if _, err := s.requireProcessTx(ctx, tx, processID); err != nil {
- return err
- }
- tasks := s.tasks.WithTx(tx)
- task, err := s.requireTask(ctx, tasks, processID, taskID)
- if err != nil {
- return err
- }
-
- s.apply(task, req)
- if err := tasks.Save(ctx, task); err != nil {
- return err
- }
- out = task
- return nil
- })
- if err != nil {
- return nil, err
- }
- return out, nil
-}
-
-// Delete removes a task from a process.
-func (s *TaskService[T, Req]) Delete(ctx context.Context, processID, taskID string) error {
- if _, err := s.requireProcess(ctx, processID); err != nil {
- return err
- }
- task, err := s.requireTask(ctx, s.tasks, processID, taskID)
- if err != nil {
- return err
- }
- return s.tasks.Delete(ctx, task)
-}
-
-// requireProcess loads the process and checks the caller may act on it: its owner, or
-// an admin. A process nobody can name is a 404; one that is not theirs is a 403.
-func (s *TaskService[T, Req]) requireProcess(ctx context.Context, processID string) (*model.BatchJobProcess, error) {
- return s.requireProcessWith(ctx, s.processes, processID)
-}
-
-func (s *TaskService[T, Req]) requireProcessTx(ctx context.Context, tx *gorm.DB, processID string) (*model.BatchJobProcess, error) {
- return s.requireProcessWith(ctx, s.processes.WithTx(tx), processID)
-}
-
-func (s *TaskService[T, Req]) requireProcessWith(ctx context.Context, processes *repository.ProcessRepository, processID string) (*model.BatchJobProcess, error) {
- principal, err := auth.RequireAuthenticated(ctx)
- if err != nil {
- return nil, err
- }
- proc, err := processes.FindByID(ctx, processID)
- if err != nil {
- return nil, notFoundAs(err, "Process not found: %s", processID)
- }
- if !proc.OwnedBy(principal.Name) && !principal.IsAdmin() {
- return nil, httpx.Forbidden("Access denied: you may only work with tasks of your own processes")
- }
- return proc, nil
-}
-
-func (s *TaskService[T, Req]) requireTask(ctx context.Context, tasks *repository.TaskRepository[T], processID, taskID string) (*T, error) {
- task, err := tasks.FindByIDAndProcessID(ctx, taskID, processID)
- if err != nil {
- return nil, notFoundAs(err, "%s not found: %s in process %s", s.name, taskID, processID)
- }
- return task, nil
-}
-
-// The four instantiations.
-type (
- DataStagingTaskService = TaskService[model.DataStagingTask, dto.DataStagingTaskRequest]
- JobSubmissionTaskService = TaskService[model.JobSubmissionTask, dto.JobSubmissionTaskRequest]
- JobMonitoringTaskService = TaskService[model.JobMonitoringTask, dto.JobMonitoringTaskRequest]
- InteractiveCommandTaskService = TaskService[model.InteractiveCommandTask, dto.InteractiveCommandTaskRequest]
-)
-
-// NewDataStagingTaskService returns a staging task service.
-func NewDataStagingTaskService(db *gorm.DB, tasks *repository.DataStagingTaskRepository, processes *repository.ProcessRepository) *DataStagingTaskService {
- return &DataStagingTaskService{
- db: db, tasks: tasks, processes: processes,
- name: "Data staging task",
- apply: dto.ApplyDataStagingTaskRequest,
- stamp: func(t *model.DataStagingTask, processID string) {
- t.ProcessID, t.ProcessType = &processID, ptrTo(model.ProcessTypeBatchJob)
- },
- }
-}
-
-// NewJobSubmissionTaskService returns a submission task service.
-func NewJobSubmissionTaskService(db *gorm.DB, tasks *repository.JobSubmissionTaskRepository, processes *repository.ProcessRepository) *JobSubmissionTaskService {
- return &JobSubmissionTaskService{
- db: db, tasks: tasks, processes: processes,
- name: "Job submission task",
- apply: dto.ApplyJobSubmissionTaskRequest,
- stamp: func(t *model.JobSubmissionTask, processID string) {
- t.ProcessID, t.ProcessType = &processID, ptrTo(model.ProcessTypeBatchJob)
- },
- }
-}
-
-// NewJobMonitoringTaskService returns a monitoring task service.
-func NewJobMonitoringTaskService(db *gorm.DB, tasks *repository.JobMonitoringTaskRepository, processes *repository.ProcessRepository) *JobMonitoringTaskService {
- return &JobMonitoringTaskService{
- db: db, tasks: tasks, processes: processes,
- name: "Job monitoring task",
- apply: dto.ApplyJobMonitoringTaskRequest,
- stamp: func(t *model.JobMonitoringTask, processID string) {
- t.ProcessID, t.ProcessType = &processID, ptrTo(model.ProcessTypeBatchJob)
- },
- }
-}
-
-// NewInteractiveCommandTaskService returns a command task service.
-func NewInteractiveCommandTaskService(db *gorm.DB, tasks *repository.InteractiveCommandTaskRepository, processes *repository.ProcessRepository) *InteractiveCommandTaskService {
- return &InteractiveCommandTaskService{
- db: db, tasks: tasks, processes: processes,
- name: "Interactive command task",
- apply: dto.ApplyInteractiveCommandTaskRequest,
- stamp: func(t *model.InteractiveCommandTask, processID string) {
- t.ProcessID, t.ProcessType = &processID, ptrTo(model.ProcessTypeBatchJob)
- },
- }
-}
-
-func ptrTo[T any](v T) *T { return &v }