refactor: introduce check state handling pipeline

Author: pixel365

cmd/pulse/main.go

@@ -9,6 +9,9 @@ import (
 
 	"github.com/joho/godotenv"
 
+	checkrepo "github.com/pixel365/pulse/internal/repository/check"
+	checksvc "github.com/pixel365/pulse/internal/services/check"
+
 	"github.com/pixel365/pulse/internal/app"
 
 	"github.com/pixel365/pulse/internal/config"
@@ -24,7 +27,11 @@ func main() {
 
 	cfg := config.MustLoad()
 
-	runner := app.NewApp(cfg)
+	repo := checkrepo.NewStateRepository()
+	stateSvc := checksvc.NewStateService(repo)
+	checkHandlerSvc := checksvc.NewHandlerService(stateSvc)
+
+	runner := app.NewApp(cfg, checkHandlerSvc)
 	if err := runner.Run(ctx); err != nil {
 		stop()
 		log.Fatalf("app run error: %v", err)

internal/app/app.go

@@ -5,6 +5,8 @@ import (
 
 	"golang.org/x/sync/errgroup"
 
+	checksvc "github.com/pixel365/pulse/internal/services/check"
+
 	"github.com/pixel365/pulse/internal"
 
 	"github.com/pixel365/pulse/internal/checker/dns"
@@ -18,52 +20,54 @@ import (
 var _ internal.Runner = (*App)(nil)
 
 type App struct {
-	cfg *config.Config
+	cfg             *config.Config
+	checkHandlerSvc checksvc.CheckHandlerService
 }
 
-func NewApp(cfg *config.Config) *App {
-	return &App{cfg}
+func NewApp(cfg *config.Config, checkSvc checksvc.CheckHandlerService) *App {
+	return &App{
+		cfg:             cfg,
+		checkHandlerSvc: checkSvc,
+	}
 }
 
 func (a *App) Run(ctx context.Context) error {
 	g, ctx := errgroup.WithContext(ctx)
 
-	w := internal.FakeWriter{}
-
 	for i := range a.cfg.HttpChecks {
-		executor := internal.NewCheckExecutor(w, a.cfg.HttpChecks[i].CheckFields)
+		executor := internal.NewCheckExecutor(a.checkHandlerSvc, a.cfg.HttpChecks[i].CheckFields)
 		checker := http.NewChecker(a.cfg.HttpChecks[i], executor)
 		g.Go(func() error {
 			return checker.Check(ctx)
 		})
 	}
 
 	for i := range a.cfg.TCPChecks {
-		executor := internal.NewCheckExecutor(w, a.cfg.TCPChecks[i].CheckFields)
+		executor := internal.NewCheckExecutor(a.checkHandlerSvc, a.cfg.TCPChecks[i].CheckFields)
 		checker := tcp.NewChecker(a.cfg.TCPChecks[i], executor)
 		g.Go(func() error {
 			return checker.Check(ctx)
 		})
 	}
 
 	for i := range a.cfg.GRPCChecks {
-		executor := internal.NewCheckExecutor(w, a.cfg.GRPCChecks[i].CheckFields)
+		executor := internal.NewCheckExecutor(a.checkHandlerSvc, a.cfg.GRPCChecks[i].CheckFields)
 		checker := grpc.NewChecker(a.cfg.GRPCChecks[i], executor)
 		g.Go(func() error {
 			return checker.Check(ctx)
 		})
 	}
 
 	for i := range a.cfg.DNSChecks {
-		executor := internal.NewCheckExecutor(w, a.cfg.DNSChecks[i].CheckFields)
+		executor := internal.NewCheckExecutor(a.checkHandlerSvc, a.cfg.DNSChecks[i].CheckFields)
 		checker := dns.NewChecker(a.cfg.DNSChecks[i], executor)
 		g.Go(func() error {
 			return checker.Check(ctx)
 		})
 	}
 
 	for i := range a.cfg.TLSChecks {
-		executor := internal.NewCheckExecutor(w, a.cfg.TLSChecks[i].CheckFields)
+		executor := internal.NewCheckExecutor(a.checkHandlerSvc, a.cfg.TLSChecks[i].CheckFields)
 		checker := tls.NewChecker(a.cfg.TLSChecks[i], executor)
 		g.Go(func() error {
 			return checker.Check(ctx)

internal/executor.go

@@ -8,6 +8,7 @@ import (
 	"github.com/pixel365/pulse/internal/config"
 	"github.com/pixel365/pulse/internal/e"
 	"github.com/pixel365/pulse/internal/model"
+	"github.com/pixel365/pulse/internal/services/check"
 )
 
 type CheckExecutor interface {
@@ -17,8 +18,8 @@ type CheckExecutor interface {
 var _ CheckExecutor = (*CheckExec)(nil)
 
 type CheckExec struct {
-	writer ResultWriter
-	cfg    config.CheckFields
+	handler check.CheckHandlerService
+	cfg     config.CheckFields
 }
 
 func (c *CheckExec) Execute(
@@ -37,8 +38,17 @@ func (c *CheckExec) Execute(
 		case <-ticker.C:
 			Sleep(ctx, c.cfg.Jitter)
 			result := c.execute(ctx, request)
+
+			policy := model.CheckPolicy{
+				CheckID:          result.CheckID,
+				ServiceID:        result.ServiceID,
+				CheckType:        result.CheckType,
+				FailureThreshold: c.cfg.FailureThreshold,
+				SuccessThreshold: c.cfg.SuccessThreshold,
+			}
+
 			//nolint:staticcheck
-			if err := c.writer.Write(ctx, result); err != nil {
+			if err := c.handler.Handle(ctx, policy, result); err != nil {
 				//TODO: log
 			}
 		}
@@ -56,7 +66,7 @@ func (c *CheckExec) execute(
 		CheckID:      c.cfg.ID,
 		ServiceID:    c.cfg.Service,
 		CheckType:    c.cfg.Type,
-		Status:       model.Success,
+		Status:       model.CheckExecutionSuccess,
 		StartedAt:    time.Now().UTC(),
 		ErrorKind:    e.ErrNone,
 		ErrorMessage: "",
@@ -84,19 +94,19 @@ func (c *CheckExec) execute(
 	result.Duration = result.FinishedAt.Sub(result.StartedAt)
 
 	if err != nil {
-		result.Status = model.Failure
+		result.Status = model.CheckExecutionFailure
 		result.ErrorKind, result.ErrorMessage = e.ResolveError(err)
 	}
 
 	return result
 }
 
 func NewCheckExecutor(
-	w ResultWriter,
+	handler check.CheckHandlerService,
 	cfg config.CheckFields,
 ) *CheckExec {
 	return &CheckExec{
-		writer: w,
-		cfg:    cfg,
+		handler: handler,
+		cfg:     cfg,
 	}
 }

internal/model/result.go

@@ -8,10 +8,21 @@ import (
 )
 
 type CheckExecutionStatus string
+type CheckStateStatus string
+type ServiceStateStatus string
 
 const (
-	Success CheckExecutionStatus = "success"
-	Failure CheckExecutionStatus = "failure"
+	CheckExecutionSuccess CheckExecutionStatus = "success"
+	CheckExecutionFailure CheckExecutionStatus = "failure"
+
+	CheckStateUnknown   CheckStateStatus = "unknown"
+	CheckStateHealthy   CheckStateStatus = "healthy"
+	CheckStateUnhealthy CheckStateStatus = "unhealthy"
+
+	ServiceStateUnknown   ServiceStateStatus = "unknown"
+	ServiceStateHealthy   ServiceStateStatus = "healthy"
+	ServiceStateUnhealthy ServiceStateStatus = "unhealthy"
+	ServiceStateDegraded  ServiceStateStatus = "degraded"
 )
 
 type CheckExecutionResult struct {
@@ -28,3 +39,29 @@ type CheckExecutionResult struct {
 	Duration      time.Duration
 	AttemptsTotal int
 }
+
+type CheckState struct {
+	UpdatedAt            time.Time
+	LastFailureAt        *time.Time
+	LastSuccessAt        *time.Time
+	LastDetails          map[string]any
+	LastStatus           CheckExecutionStatus
+	LastExecutionID      string
+	CheckID              string
+	LastErrorKind        e.ErrorKind
+	LastErrorMessage     string
+	Status               CheckStateStatus
+	CheckType            config.CheckType
+	ServiceID            string
+	LastDuration         time.Duration
+	ConsecutiveSuccesses int
+	ConsecutiveFailures  int
+}
+
+type CheckPolicy struct {
+	CheckID          string
+	ServiceID        string
+	CheckType        config.CheckType
+	FailureThreshold int
+	SuccessThreshold int
+}

internal/repository/check/interfaces.go

@@ -0,0 +1,12 @@
+package check
+
+import (
+	"context"
+
+	"github.com/pixel365/pulse/internal/model"
+)
+
+type CheckStateRepository interface {
+	GetCheckState(context.Context, string) (*model.CheckState, error)
+	UpdateCheckState(context.Context, model.CheckState) error
+}

internal/repository/check/state.go

@@ -0,0 +1,24 @@
+package check
+
+import (
+	"context"
+
+	"github.com/pixel365/pulse/internal/model"
+)
+
+var _ CheckStateRepository = (*StateCheck)(nil)
+
+type StateCheck struct {
+}
+
+func (s *StateCheck) GetCheckState(ctx context.Context, service string) (*model.CheckState, error) {
+	return nil, nil
+}
+
+func (s *StateCheck) UpdateCheckState(ctx context.Context, state model.CheckState) error {
+	return nil
+}
+
+func NewStateRepository() *StateCheck {
+	return &StateCheck{}
+}

internal/services/check/handler.go

@@ -0,0 +1,27 @@
+package check
+
+import (
+	"context"
+
+	"github.com/pixel365/pulse/internal/model"
+)
+
+var _ CheckHandlerService = (*Handler)(nil)
+
+type Handler struct {
+	svc CheckStateService
+}
+
+func (h *Handler) Handle(
+	ctx context.Context,
+	policy model.CheckPolicy,
+	result model.CheckExecutionResult,
+) error {
+	return nil
+}
+
+func NewHandlerService(svc CheckStateService) *Handler {
+	return &Handler{
+		svc: svc,
+	}
+}

internal/services/check/interfaces.go

@@ -0,0 +1,16 @@
+package check
+
+import (
+	"context"
+
+	"github.com/pixel365/pulse/internal/model"
+)
+
+type CheckHandlerService interface {
+	Handle(context.Context, model.CheckPolicy, model.CheckExecutionResult) error
+}
+
+type CheckStateService interface {
+	GetState(context.Context, string) (*model.CheckState, error)
+	UpsertState(context.Context, model.CheckState) error
+}

internal/services/check/state.go

@@ -0,0 +1,30 @@
+package check
+
+import (
+	"context"
+
+	"github.com/pixel365/pulse/internal/model"
+	"github.com/pixel365/pulse/internal/repository/check"
+)
+
+var _ CheckStateService = (*State)(nil)
+
+type State struct {
+	repo check.CheckStateRepository
+}
+
+func (s *State) GetState(ctx context.Context, service string) (*model.CheckState, error) {
+	result, err := s.repo.GetCheckState(ctx, service)
+
+	return result, err
+}
+
+func (s *State) UpsertState(ctx context.Context, state model.CheckState) error {
+	err := s.repo.UpdateCheckState(ctx, state)
+
+	return err
+}
+
+func NewStateService(repo check.CheckStateRepository) *State {
+	return &State{repo: repo}
+}