process visibility task

engine/immediate_task_concurrent_processor.go

@@ -158,7 +158,18 @@
 
 func (w *immediateTaskConcurrentProcessor) processVisibilityTask(
 	ctx context.Context, task data_models.ImmediateTask) error {
-	return nil
+	if task.ImmediateTaskInfo.VisibilityInfo == nil {
+		return fmt.Errorf("visibility info is not set")
+	}
+	return w.visibilityStore.RecordProcessExecutionStatus(ctx, data_models.RecordProcessExecutionStatusRequest{
+		Namespace:          task.ImmediateTaskInfo.VisibilityInfo.Namespace,
+		ProcessId:          task.ImmediateTaskInfo.VisibilityInfo.ProcessId,
+		ProcessExecutionId: task.ProcessExecutionId,
+		ProcessType:        task.ImmediateTaskInfo.VisibilityInfo.ProcessType,
+		Status:             task.ImmediateTaskInfo.VisibilityInfo.Status,
+		StartTime:          task.ImmediateTaskInfo.VisibilityInfo.StartTime,
+		CloseTime:          task.ImmediateTaskInfo.VisibilityInfo.CloseTime,
+	})
 }
 
 func (w *immediateTaskConcurrentProcessor) processWaitUntilTask(

extensions/data_models_row.go

@@ -257,4 +257,15 @@ type (
 		Key                      string
 		Value                    types.JSONText
 	}
+
+	ExecutionVisibilityRow struct {
+		Namespace                string
+		ProcessId                string
+		ProcessExecutionId       uuid.UUID
+		ProcessExecutionIdString string
+		ProcessTypeName          string
+		Status                   data_models.ProcessExecutionStatus
+		StartTime                time.Time
+		CloseTime                time.Time
+	}
 )

extensions/postgres/non_transactional.go

@@ -202,3 +202,30 @@ func (d dbSession) SelectLocalAttributes(
 	err = d.db.SelectContext(ctx, &rows, query, args...)
 	return rows, err
 }
+
+const insertProcessExecutionStartQuery = `INSERT INTO xcherry_sys_executions_visibility
+	(namespace, process_id, process_execution_id, process_type_name, status, start_time)
+	VALUES (:namespace, :process_id, :process_execution_id_string, :process_type_name, :status, :start_time)`
+
+func (d dbSession) InsertProcessExecutionStartForVisibility(
+	ctx context.Context, row extensions.ExecutionVisibilityRow,
+) error {
+	row.StartTime = ToPostgresDateTime(row.StartTime)
+	row.ProcessExecutionIdString = row.ProcessExecutionId.String()
+	_, err := d.db.NamedExecContext(ctx, insertProcessExecutionStartQuery, row)
+	return err
+}
+
+const updateProcessExecutionStatusQuery = `UPDATE xcherry_sys_executions_visibility
+	SET status = :status, close_time = :close_time
+	WHERE namespace = :namespace AND process_execution_id = :process_execution_id_string
+`
+
+func (d dbSession) UpdateProcessExecutionStatusForVisibility(
+	ctx context.Context, row extensions.ExecutionVisibilityRow,
+) error {
+	row.CloseTime = ToPostgresDateTime(row.CloseTime)
+	row.ProcessExecutionIdString = row.ProcessExecutionId.String()
+	_, err := d.db.NamedExecContext(ctx, updateProcessExecutionStatusQuery, row)
+	return err
+}

extensions/postgres/schema/xcherry_sys_schema.sql

@@ -81,7 +81,7 @@ CREATE TABLE xcherry_sys_local_attributes(
     PRIMARY KEY (process_execution_id, key)
 );
 
-CREATE TABLE xcherry_executions_visibility (
+CREATE TABLE xcherry_sys_executions_visibility (
     namespace VARCHAR(31) NOT NULL,
     process_id VARCHAR(255) NOT NULL,
     process_execution_id uuid NOT NULL,
@@ -92,12 +92,12 @@ CREATE TABLE xcherry_executions_visibility (
     PRIMARY KEY (namespace, process_execution_id)
 );
 
-CREATE INDEX by_start_time ON xcherry_executions_visibility (namespace, start_time DESC, process_execution_id);
+CREATE INDEX by_start_time ON xcherry_sys_executions_visibility (namespace, start_time DESC, process_execution_id);
 
-CREATE INDEX by_type_start_time ON xcherry_executions_visibility (namespace, process_type_name, start_time DESC, process_execution_id);
+CREATE INDEX by_type_start_time ON xcherry_sys_executions_visibility (namespace, process_type_name, start_time DESC, process_execution_id);
 
-CREATE INDEX by_process_id_start_time ON xcherry_executions_visibility (namespace, process_id, start_time DESC, process_execution_id);
+CREATE INDEX by_process_id_start_time ON xcherry_sys_executions_visibility (namespace, process_id, start_time DESC, process_execution_id);
 
-CREATE INDEX by_status_start_time ON xcherry_executions_visibility (namespace, status, start_time DESC, process_execution_id);
+CREATE INDEX by_status_start_time ON xcherry_sys_executions_visibility (namespace, status, start_time DESC, process_execution_id);
 
-CREATE INDEX by_status_type_start_time ON xcherry_executions_visibility (namespace, status, process_type_name, start_time DESC, process_execution_id);
+CREATE INDEX by_status_type_start_time ON xcherry_sys_executions_visibility (namespace, status, process_type_name, start_time DESC, process_execution_id);

extensions/sql_db_interfaces.go

@@ -106,6 +106,14 @@ type nonTransactionalCRUD interface {
 	SelectLocalAttributes(
 		ctx context.Context, processExecutionId uuid.UUID, keys []string,
 	) ([]LocalAttributeRow, error)
+
+	InsertProcessExecutionStartForVisibility(
+		ctx context.Context, row ExecutionVisibilityRow,
+	) error
+
+	UpdateProcessExecutionStatusForVisibility(
+		ctx context.Context, row ExecutionVisibilityRow,
+	) error
 }
 
 type ErrorChecker interface {

persistence/data_models/record_process_execution_status_req.go

@@ -11,6 +11,6 @@ type RecordProcessExecutionStatusRequest struct {
 	ProcessExecutionId uuid.UUID
 	ProcessType        string
 	Status             ProcessExecutionStatus
-	StartTime          int64
-	EndTime            int64
+	StartTime          *int64
+	CloseTime          *int64
 }

persistence/data_models/visibility_info_json.go

@@ -3,11 +3,14 @@
 
 package data_models
 
+import "github.com/xcherryio/xcherry/common/uuid"
+
 type VisibilityInfoJson struct {
-	Namespace   string
-	ProcessId   string
-	ProcessType string
-	Status      ProcessExecutionStatus
-	StartTime   int64
-	EndTime     int64
+	Namespace          string                 `json:"namespace"`
+	ProcessId          string                 `json:"processId"`
+	ProcessExecutionId uuid.UUID              `json:"processExecutionId"`
+	ProcessType        string                 `json:"processType"`
+	Status             ProcessExecutionStatus `json:"status"`
+	StartTime          *int64                 `json:"startTime"`
+	CloseTime          *int64                 `json:"closeTime"`
 }

persistence/process/complete_execute.go

@@ -6,6 +6,7 @@ package process
 import (
 	"context"
 	"fmt"
+	"github.com/xcherryio/xcherry/common/ptr"
 	"time"
 
 	"github.com/xcherryio/xcherry/persistence/data_models"
@@ -166,8 +167,8 @@ func (p sqlProcessStoreImpl) doCompleteExecuteExecutionTx(
 			request.Prepare.Info.ProcessType,
 			request.ProcessExecutionId,
 			prcRow.Status,
-			-1,
-			time.Now().Unix(),
+			nil,
+			ptr.Any(time.Now().Unix()),
 		)
 		if err != nil {
 			return nil, err

persistence/process/record_status.go

@@ -19,17 +19,18 @@ func (p sqlProcessStoreImpl) AddVisibilityTaskRecordProcessExecutionStatus(
 	processType string,
 	processExecutionId uuid.UUID,
 	status data_models.ProcessExecutionStatus,
-	startTime int64,
-	endTime int64) error {
+	startTime *int64,
+	endTime *int64) error {
 
 	visibilityTaskInfo := data_models.ImmediateTaskInfoJson{
 		VisibilityInfo: &data_models.VisibilityInfoJson{
-			Namespace:   namespace,
-			ProcessId:   processId,
-			ProcessType: processType,
-			Status:      status,
-			StartTime:   startTime,
-			EndTime:     endTime,
+			Namespace:          namespace,
+			ProcessId:          processId,
+			ProcessType:        processType,
+			ProcessExecutionId: processExecutionId,
+			Status:             status,
+			StartTime:          startTime,
+			CloseTime:          endTime,
 		},
 	}
 

persistence/process/sqltest/base_basic.go

@@ -48,12 +48,13 @@ func SQLBasicTest(t *testing.T, ass *assert.Assertions, store persistence.Proces
 		"-0",
 		data_models.ImmediateTaskInfoJson{
 			VisibilityInfo: &data_models.VisibilityInfoJson{
-				Namespace:   namespace,
-				ProcessId:   processId,
-				ProcessType: testProcessType,
-				Status:      data_models.ProcessExecutionStatusRunning,
-				StartTime:   -1,
-				EndTime:     -1,
+				Namespace:          namespace,
+				ProcessId:          processId,
+				ProcessType:        testProcessType,
+				ProcessExecutionId: prcExeId,
+				Status:             data_models.ProcessExecutionStatusRunning,
+				StartTime:          nil,
+				CloseTime:          nil,
 			},
 		})
 

persistence/process/start_process.go

@@ -6,6 +6,7 @@ package process
 import (
 	"context"
 	"fmt"
+	"github.com/xcherryio/xcherry/common/ptr"
 	"time"
 
 	"github.com/xcherryio/apis/goapi/xcapi"
@@ -318,8 +319,8 @@ func (p sqlProcessStoreImpl) applyTerminateIfRunningPolicy(
 				request.Request.ProcessType,
 				processExecutionRowForUpdate.ProcessExecutionId,
 				data_models.ProcessExecutionStatusTerminated,
-				-1,
-				time.Now().Unix())
+				nil,
+				ptr.Any(time.Now().Unix()))
 			if err != nil {
 				return nil, err
 			}
@@ -518,8 +519,8 @@ func (p sqlProcessStoreImpl) insertProcessExecution(
 		request.Request.ProcessType,
 		processExecutionId,
 		data_models.ProcessExecutionStatusRunning,
-		startTime.Unix(),
-		-1)
+		ptr.Any(startTime.Unix()),
+		nil)
 	if err != nil {
 		return hasNewImmediateTask, err
 	}

persistence/process/stop_process.go

@@ -6,6 +6,7 @@ package process
 import (
 	"context"
 	"github.com/xcherryio/apis/goapi/xcapi"
+	"github.com/xcherryio/xcherry/common/ptr"
 	"github.com/xcherryio/xcherry/persistence/data_models"
 	"time"
 
@@ -116,8 +117,8 @@ func (p sqlProcessStoreImpl) doStopProcessTx(
 		procExecInfoJson.ProcessType,
 		curProcExecRow.ProcessExecutionId,
 		status,
-		-1,
-		time.Now().Unix(),
+		nil,
+		ptr.Any(time.Now().Unix()),
 	)
 	if err != nil {
 		return nil, err

persistence/visibility/visibility_store.go

@@ -5,11 +5,13 @@
 
 import (
 	"context"
+	"fmt"
 	"github.com/xcherryio/xcherry/common/log"
 	"github.com/xcherryio/xcherry/config"
 	"github.com/xcherryio/xcherry/extensions"
 	"github.com/xcherryio/xcherry/persistence"
 	"github.com/xcherryio/xcherry/persistence/data_models"
+	"time"
 )
 
 type sqlVisibilityStoreImpl struct {
@@ -31,6 +33,29 @@
 
 func (p sqlVisibilityStoreImpl) RecordProcessExecutionStatus(
 	ctx context.Context, req data_models.RecordProcessExecutionStatusRequest) error {
-	// TODO: add implementation
-	return nil
+	if req.Status == data_models.ProcessExecutionStatusUndefined {
+		return fmt.Errorf("process status is undefined")
+	}
+
+	if req.Status == data_models.ProcessExecutionStatusRunning {
+		if req.StartTime == nil {
+			return fmt.Errorf("start time is required for recording visibility for running process")
+		}
+		return p.session.InsertProcessExecutionStartForVisibility(ctx, extensions.ExecutionVisibilityRow{
+			Namespace:          req.Namespace,
+			ProcessId:          req.ProcessId,
+			ProcessExecutionId: req.ProcessExecutionId,
+			ProcessTypeName:    req.ProcessType,
+			Status:             req.Status,
+			StartTime:          time.Unix(*req.StartTime, 0),
+		})
+	}
+	return p.session.UpdateProcessExecutionStatusForVisibility(ctx, extensions.ExecutionVisibilityRow{
+		Namespace:          req.Namespace,
+		ProcessId:          req.ProcessId,
+		ProcessExecutionId: req.ProcessExecutionId,
+		ProcessTypeName:    req.ProcessType,
+		Status:             req.Status,
+		CloseTime:          time.Unix(*req.CloseTime, 0),
+	})
 }