Add WaitForWorkflowInstance method

Author: cschleiden

backend/backend.go

@@ -8,6 +8,13 @@ import (
 	"github.com/cschleiden/go-workflows/workflow"
 )
 
+type WorkflowState int
+
+const (
+	WorkflowStateActive WorkflowState = iota
+	WorkflowStateFinished
+)
+
 //go:generate mockery --name=Backend --inpackage
 type Backend interface {
 	// CreateWorkflowInstance creates a new workflow instance
@@ -16,6 +23,12 @@ type Backend interface {
 	// CancelWorkflowInstance cancels a running workflow instance
 	CancelWorkflowInstance(ctx context.Context, instance workflow.Instance) error
 
+	// GetWorkflowInstanceState returns the state of the given workflow instance
+	GetWorkflowInstanceState(ctx context.Context, instance workflow.Instance) (WorkflowState, error)
+
+	// GetWorkflowInstanceHistory returns the full workflow history for the given instance
+	GetWorkflowInstanceHistory(ctx context.Context, instance workflow.Instance) ([]history.Event, error)
+
 	// SignalWorkflow signals a running workflow instance
 	SignalWorkflow(ctx context.Context, instanceID string, event history.Event) error
 

backend/mock_Backend.go

@@ -120,6 +120,77 @@ func (b *mysqlBackend) CancelWorkflowInstance(ctx context.Context, instance work
 	return tx.Commit()
 }
 
+func (b *mysqlBackend) GetWorkflowInstanceHistory(ctx context.Context, instance workflow.Instance) ([]history.Event, error) {
+	tx, err := b.db.BeginTx(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	historyEvents, err := tx.QueryContext(
+		ctx,
+		"SELECT event_id, instance_id, event_type, timestamp, schedule_event_id, attributes, visible_at FROM `history` WHERE instance_id = ? ORDER BY id",
+		instance.GetInstanceID(),
+	)
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get history")
+	}
+
+	h := make([]history.Event, 0)
+
+	for historyEvents.Next() {
+		var instanceID string
+		var attributes []byte
+
+		historyEvent := history.Event{}
+
+		if err := historyEvents.Scan(
+			&historyEvent.ID,
+			&instanceID,
+			&historyEvent.Type,
+			&historyEvent.Timestamp,
+			&historyEvent.ScheduleEventID,
+			&attributes,
+			&historyEvent.VisibleAt,
+		); err != nil {
+			return nil, errors.Wrap(err, "could not scan event")
+		}
+
+		a, err := history.DeserializeAttributes(historyEvent.Type, attributes)
+		if err != nil {
+			return nil, errors.Wrap(err, "could not deserialize attributes")
+		}
+
+		historyEvent.Attributes = a
+
+		h = append(h, historyEvent)
+	}
+
+	return h, nil
+}
+
+func (b *mysqlBackend) GetWorkflowInstanceState(ctx context.Context, instance workflow.Instance) (backend.WorkflowState, error) {
+	row := b.db.QueryRowContext(
+		ctx,
+		"SELECT completed_at FROM instances WHERE instance_id = ? AND execution_id = ?",
+		instance.GetInstanceID(),
+		instance.GetExecutionID(),
+	)
+
+	var completedAt sql.NullTime
+	if err := row.Scan(&completedAt); err != nil {
+		if err == sql.ErrNoRows {
+			return backend.WorkflowStateActive, errors.New("could not find workflow instance")
+		}
+	}
+
+	if completedAt.Valid {
+		return backend.WorkflowStateFinished, nil
+	}
+
+	return backend.WorkflowStateActive, nil
+}
+
 func createInstance(ctx context.Context, tx *sql.Tx, wfi workflow.Instance) error {
 	var parentInstanceID *string
 	var parentEventID *int

backend/mysql/mysql.go

@@ -8,7 +8,6 @@ CREATE TABLE IF NOT EXISTS `instances` (
   `locked_until` DATETIME NULL,
   `sticky_until` DATETIME NULL,
   `worker` TEXT NULL
-
 );
 
 CREATE INDEX IF NOT EXISTS `idx_instances_locked_until_completed_at` ON `instances` (`locked_until`, `sticky_until`, `completed_at`, `worker`);

backend/sqlite/schema.sql

@@ -146,6 +146,43 @@ func (sb *sqliteBackend) CancelWorkflowInstance(ctx context.Context, instance wo
 	return tx.Commit()
 }
 
+func (sb *sqliteBackend) GetWorkflowInstanceHistory(ctx context.Context, instance workflow.Instance) ([]history.Event, error) {
+	tx, err := sb.db.BeginTx(ctx, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer tx.Rollback()
+
+	h, err := getHistory(ctx, tx, instance.GetInstanceID())
+	if err != nil {
+		return nil, errors.Wrap(err, "could not get workflow history")
+	}
+
+	return h, nil
+}
+
+func (s *sqliteBackend) GetWorkflowInstanceState(ctx context.Context, instance workflow.Instance) (backend.WorkflowState, error) {
+	row := s.db.QueryRowContext(
+		ctx,
+		"SELECT completed_at FROM instances WHERE id = ? AND execution_id = ?",
+		instance.GetInstanceID(),
+		instance.GetExecutionID(),
+	)
+
+	var completedAt sql.NullTime
+	if err := row.Scan(&completedAt); err != nil {
+		if err == sql.ErrNoRows {
+			return backend.WorkflowStateActive, errors.New("could not find workflow instance")
+		}
+	}
+
+	if completedAt.Valid {
+		return backend.WorkflowStateFinished, nil
+	}
+
+	return backend.WorkflowStateActive, nil
+}
+
 func (sb *sqliteBackend) SignalWorkflow(ctx context.Context, instanceID string, event history.Event) error {
 	tx, err := sb.db.BeginTx(ctx, nil)
 	if err != nil {

backend/sqlite/sqlite.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"time"
 
+	"github.com/benbjohnson/clock"
 	"github.com/cschleiden/go-workflows/backend"
 	a "github.com/cschleiden/go-workflows/internal/args"
 	"github.com/cschleiden/go-workflows/internal/converter"
@@ -15,6 +16,9 @@ import (
 	"github.com/pkg/errors"
 )
 
+var ErrWorkflowCanceled = errors.New("workflow canceled")
+var ErrWorkflowTerminated = errors.New("workflow terminated")
+
 type WorkflowInstanceOptions struct {
 	InstanceID string
 }
@@ -24,16 +28,20 @@ type Client interface {
 
 	CancelWorkflowInstance(ctx context.Context, instance workflow.Instance) error
 
+	WaitForWorkflowInstance(ctx context.Context, instance workflow.Instance, timeout time.Duration) error
+
 	SignalWorkflow(ctx context.Context, instanceID string, name string, arg interface{}) error
 }
 
 type client struct {
 	backend backend.Backend
+	clock   clock.Clock
 }
 
 func New(backend backend.Backend) Client {
 	return &client{
 		backend: backend,
+		clock:   clock.New(),
 	}
 }
 
@@ -44,7 +52,7 @@ func (c *client) CreateWorkflowInstance(ctx context.Context, options WorkflowIns
 	}
 
 	startedEvent := history.NewHistoryEvent(
-		time.Now(),
+		c.clock.Now(),
 		history.EventType_WorkflowExecutionStarted,
 		&history.ExecutionStartedAttributes{
 			Name:   fn.Name(wf),
@@ -76,7 +84,7 @@ func (c *client) SignalWorkflow(ctx context.Context, instanceID string, name str
 	}
 
 	event := history.NewHistoryEvent(
-		time.Now(),
+		c.clock.Now(),
 		history.EventType_SignalReceived,
 		&history.SignalReceivedAttributes{
 			Name: name,
@@ -86,3 +94,79 @@ func (c *client) SignalWorkflow(ctx context.Context, instanceID string, name str
 
 	return c.backend.SignalWorkflow(ctx, instanceID, event)
 }
+
+func (c *client) WaitForWorkflowInstance(ctx context.Context, instance workflow.Instance, timeout time.Duration) error {
+	if timeout == 0 {
+		timeout = time.Second * 20
+	}
+
+	ticker := c.clock.Ticker(time.Second)
+	defer ticker.Stop()
+
+	ctx, cancel := c.clock.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	for {
+		s, err := c.backend.GetWorkflowInstanceState(ctx, instance)
+		if err != nil {
+			return errors.Wrap(err, "could not get workflow state")
+		}
+
+		if s == backend.WorkflowStateFinished {
+			return nil
+		}
+
+		ticker.Reset(time.Second)
+		select {
+		case <-ticker.C:
+			continue
+
+		case <-ctx.Done():
+			return errors.New("workflow did not finish in specified timeout")
+		}
+	}
+}
+
+func GetWorkflowResult[T any](ctx context.Context, c Client, instance workflow.Instance, timeout time.Duration) (T, error) {
+	// Zero result
+	var z T
+
+	if err := c.WaitForWorkflowInstance(ctx, instance, timeout); err != nil {
+		return z, errors.Wrap(err, "workflow did not finish in time")
+	}
+
+	ic := c.(*client)
+	b := ic.backend
+
+	h, err := b.GetWorkflowInstanceHistory(ctx, instance)
+	if err != nil {
+		return z, errors.Wrap(err, "could not get workflow history")
+	}
+
+	// Iterate over history backwards
+	for i := len(h) - 1; i >= 0; i-- {
+		event := h[i]
+		switch event.Type {
+		case history.EventType_WorkflowExecutionFinished:
+			a := event.Attributes.(*history.ExecutionCompletedAttributes)
+			if a.Error != "" {
+				return z, errors.New(a.Error)
+			}
+
+			var r T
+			if err := converter.DefaultConverter.From(a.Result, &r); err != nil {
+				return z, errors.Wrap(err, "could not convert result")
+			}
+
+			return r, nil
+
+		case history.EventType_WorkflowExecutionCanceled:
+			return z, ErrWorkflowCanceled
+
+		case history.EventType_WorkflowExecutionTerminated:
+			return z, ErrWorkflowTerminated
+		}
+	}
+
+	return z, errors.New("workflow finished, but could not find result event")
+}