feat: introduce notifier (#71)

Author: w-h-a

cmd/engine.go

@@ -13,6 +13,8 @@ import (
 	memorybroker "github.com/w-h-a/workflow/internal/engine/clients/broker/memory"
 	"github.com/w-h-a/workflow/internal/engine/clients/broker/nats"
 	"github.com/w-h-a/workflow/internal/engine/clients/broker/rabbit"
+	"github.com/w-h-a/workflow/internal/engine/clients/notifier"
+	"github.com/w-h-a/workflow/internal/engine/clients/notifier/local"
 	"github.com/w-h-a/workflow/internal/engine/clients/readwriter"
 	memoryreadwriter "github.com/w-h-a/workflow/internal/engine/clients/readwriter/memory"
 	"github.com/w-h-a/workflow/internal/engine/clients/readwriter/postgres"
@@ -173,10 +175,12 @@ func StartEngine(ctx *cli.Context) error {
 	var c *coordinator.Service
 	if config.Mode() == "standalone" || config.Mode() == "coordinator" {
 		readwriterClient := initReadWriter()
+		notifierClient := initNotifier()
 
 		coordinatorServer, c = engine.NewCoordinator(
 			brokerClient,
 			readwriterClient,
+			notifierClient,
 		)
 
 		numServices += 2
@@ -338,3 +342,7 @@ func initReadWriter() readwriter.ReadWriter {
 		return memoryreadwriter.NewReadWriter()
 	}
 }
+
+func initNotifier() notifier.Notifier {
+	return local.NewNotifier()
+}

internal/engine/clients/notifier/event.go

@@ -0,0 +1,15 @@
+package notifier
+
+type AlertType string
+
+const (
+	Success AlertType = "SUCCESS"
+	Failure AlertType = "FAILURE"
+)
+
+type Event struct {
+	Source  string
+	Type    AlertType
+	Title   string
+	Payload map[string]any
+}

internal/engine/clients/notifier/local/notifier.go

@@ -0,0 +1,35 @@
+package local
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/w-h-a/workflow/internal/engine/clients/notifier"
+)
+
+type localNotifier struct {
+	options notifier.Options
+}
+
+func (n *localNotifier) Notify(ctx context.Context, event notifier.Event, opts ...notifier.NotifyOption) error {
+	slog.InfoContext(
+		ctx,
+		"NOTIFY EVENT",
+		"title", event.Title,
+		"source", event.Source,
+		"type", event.Type,
+		"payload", event.Payload,
+	)
+
+	return nil
+}
+
+func NewNotifier(opts ...notifier.Option) notifier.Notifier {
+	options := notifier.NewOptions(opts...)
+
+	n := &localNotifier{
+		options: options,
+	}
+
+	return n
+}

internal/engine/clients/notifier/notifier.go

@@ -0,0 +1,19 @@
+package notifier
+
+import "context"
+
+type NotifierType string
+
+const (
+	Local NotifierType = "local"
+)
+
+var (
+	NotifierTypes = map[string]NotifierType{
+		"local": Local,
+	}
+)
+
+type Notifier interface {
+	Notify(ctx context.Context, event Event, opts ...NotifyOption) error
+}

internal/engine/clients/notifier/options.go

@@ -0,0 +1,53 @@
+package notifier
+
+import "context"
+
+type Option func(o *Options)
+
+type Options struct {
+	URL     string
+	Secret  string
+	Context context.Context
+}
+
+func WithURL(url string) Option {
+	return func(o *Options) {
+		o.URL = url
+	}
+}
+
+func WithSecret(secret string) Option {
+	return func(o *Options) {
+		o.Secret = secret
+	}
+}
+
+func NewOptions(opts ...Option) Options {
+	options := Options{
+		Context: context.Background(),
+	}
+
+	for _, fn := range opts {
+		fn(&options)
+	}
+
+	return options
+}
+
+type NotifyOption func(o *NotifyOptions)
+
+type NotifyOptions struct {
+	Context context.Context
+}
+
+func NewNotifyOptions(opts ...NotifyOption) NotifyOptions {
+	options := NotifyOptions{
+		Context: context.Background(),
+	}
+
+	for _, fn := range opts {
+		fn(&options)
+	}
+
+	return options
+}

internal/engine/config/config.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/w-h-a/workflow/internal/engine/clients/broker"
+	"github.com/w-h-a/workflow/internal/engine/clients/notifier"
 	"github.com/w-h-a/workflow/internal/engine/clients/readwriter"
 	"github.com/w-h-a/workflow/internal/engine/clients/runner"
 	"github.com/w-h-a/workflow/internal/task"
@@ -40,6 +41,7 @@ type config struct {
 	runnerPruneInterval time.Duration
 	readwriter          string
 	readwriterLocation  string
+	notifier            string
 }
 
 func New() {
@@ -65,6 +67,7 @@ func New() {
 			runnerPruneInterval: 24 * time.Hour,
 			readwriter:          "memory",
 			readwriterLocation:  "",
+			notifier:            "local",
 		}
 
 		env := os.Getenv("ENV")
@@ -201,6 +204,15 @@ func New() {
 		if len(readwriterLocation) > 0 {
 			instance.readwriterLocation = readwriterLocation
 		}
+
+		n := os.Getenv("NOTIFIER")
+		if len(n) > 0 {
+			if _, ok := notifier.NotifierTypes[n]; ok {
+				instance.notifier = n
+			} else {
+				panic("unsupported notifier")
+			}
+		}
 	})
 }
 

internal/engine/engine.go

@@ -7,6 +7,7 @@ import (
 	"github.com/w-h-a/pkg/serverv2"
 	httpserver "github.com/w-h-a/pkg/serverv2/http"
 	"github.com/w-h-a/workflow/internal/engine/clients/broker"
+	"github.com/w-h-a/workflow/internal/engine/clients/notifier"
 	"github.com/w-h-a/workflow/internal/engine/clients/readwriter"
 	"github.com/w-h-a/workflow/internal/engine/clients/runner"
 	"github.com/w-h-a/workflow/internal/engine/config"
@@ -23,11 +24,13 @@ import (
 func NewCoordinator(
 	brokerClient broker.Broker,
 	readwriterClient readwriter.ReadWriter,
+	notifierClient notifier.Notifier,
 ) (serverv2.Server, *coordinator.Service) {
 	// services
 	coordinatorService := coordinator.New(
 		brokerClient,
 		readwriterClient,
+		notifierClient,
 		map[string]int{
 			string(task.Started):   1,
 			string(task.Completed): 1,

internal/engine/services/coordinator/service.go

@@ -4,13 +4,15 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"log/slog"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/google/uuid"
 	"github.com/w-h-a/workflow/internal/engine/clients/broker"
+	"github.com/w-h-a/workflow/internal/engine/clients/notifier"
 	"github.com/w-h-a/workflow/internal/engine/clients/reader"
 	"github.com/w-h-a/workflow/internal/engine/clients/readwriter"
 	"github.com/w-h-a/workflow/internal/task"
@@ -23,6 +25,7 @@ import (
 type Service struct {
 	broker     broker.Broker
 	readwriter readwriter.ReadWriter
+	notifier   notifier.Notifier
 	queues     map[string]int
 	locks      map[string]*sync.RWMutex
 	mtx        sync.RWMutex
@@ -286,6 +289,19 @@ func (s *Service) handleTask(ctx context.Context, data []byte) error {
 
 		s.removeTaskLock(t.ID)
 
+		event := notifier.Event{
+			Source:  "workflow-coordinator",
+			Type:    notifier.Success,
+			Title:   fmt.Sprintf("Task %s Completed Successfully", t.ID),
+			Payload: map[string]any{},
+		}
+
+		if err := s.notifier.Notify(ctx, event); err != nil {
+			child.RecordError(err)
+			child.SetStatus(codes.Error, err.Error())
+			return err
+		}
+
 		child.SetStatus(codes.Ok, "task completed event handled")
 
 		return nil
@@ -301,6 +317,21 @@ func (s *Service) handleTask(ctx context.Context, data []byte) error {
 
 		s.removeTaskLock(t.ID)
 
+		event := notifier.Event{
+			Source: "workflow-coordinator",
+			Type:   notifier.Failure,
+			Title:  fmt.Sprintf("Task %s Failed", t.ID),
+			Payload: map[string]any{
+				"error_msg": t.Error,
+			},
+		}
+
+		if err := s.notifier.Notify(ctx, event); err != nil {
+			child.RecordError(err)
+			child.SetStatus(codes.Error, err.Error())
+			return err
+		}
+
 		child.SetStatus(codes.Ok, "task failed event handled")
 
 		return nil
@@ -377,10 +408,11 @@ func (s *Service) removeTaskLock(id string) {
 	delete(s.locks, id)
 }
 
-func New(b broker.Broker, rw readwriter.ReadWriter, qs map[string]int) *Service {
+func New(b broker.Broker, rw readwriter.ReadWriter, n notifier.Notifier, qs map[string]int) *Service {
 	return &Service{
 		broker:     b,
 		readwriter: rw,
+		notifier:   n,
 		queues:     qs,
 		locks:      map[string]*sync.RWMutex{},
 		mtx:        sync.RWMutex{},