feat(storage): persist raw check execution results in postgres

Author: pixel365

cmd/pulse/main.go

@@ -8,6 +8,8 @@ import (
 
 	"github.com/joho/godotenv"
 
+	"github.com/pixel365/pulse/internal/db/postgres"
+
 	"github.com/pixel365/pulse/internal/logger"
 
 	checkrepo "github.com/pixel365/pulse/internal/repository/check"
@@ -26,9 +28,19 @@ func main() {
 
 	cfg := config.MustLoad()
 	log := logger.NewSlog()
-	repo := checkrepo.NewStateRepository()
-	stateSvc := checksvc.NewStateService(repo)
-	checkHandlerSvc := checksvc.NewHandlerService(stateSvc)
+
+	pgConfig := postgres.NewConfigFromEnv()
+	pgPool, err := postgres.NewPool(ctx, pgConfig)
+	if err != nil {
+		log.Error(ctx, "postgres pool error", "error", err)
+		return
+	}
+	defer pgPool.Close()
+
+	staterepo := checkrepo.NewStateRepository(pgPool)
+	execRepo := checkrepo.NewExecutionRepository(pgPool)
+	stateSvc := checksvc.NewStateService(staterepo)
+	checkHandlerSvc := checksvc.NewHandlerService(stateSvc, execRepo)
 
 	runner := app.NewApp(cfg, log, checkHandlerSvc)
 	if err := runner.Run(ctx); err != nil {

internal/db/postgres/config.go

@@ -4,14 +4,11 @@ import (
 	"fmt"
 	"net/url"
 	"os"
-
-	"github.com/pixel365/pulse/internal/logger"
 )
 
 type ConfigOption func(*Config)
 
 type Config struct {
-	Log      logger.Logger
 	User     string
 	Password string
 	Host     string

internal/db/postgres/postgres.go

@@ -8,38 +8,24 @@ import (
 )
 
 func NewPool(ctx context.Context, cfg Config) (*pgxpool.Pool, error) {
-	log := cfg.Log
 	config, err := pgxpool.ParseConfig(cfg.DSN())
 	if err != nil {
-		if log != nil {
-			log.Error(ctx, "cannot parse config", "error", err)
-		}
 		return nil, err
 	}
 
 	config.ConnConfig.Tracer = otelpgx.NewTracer()
 
 	pool, err := pgxpool.NewWithConfig(ctx, config)
 	if err != nil {
-		if log != nil {
-			log.Error(ctx, "cannot create pool", "error", err)
-		}
 		return nil, err
 	}
 
 	if err = otelpgx.RecordStats(pool); err != nil {
-		if log != nil {
-			log.Error(ctx, "cannot record stats", "error", err)
-		}
 		pool.Close()
 		return nil, err
 	}
 
 	if err = pool.Ping(ctx); err != nil {
-		if log != nil {
-			log.Error(ctx, "cannot ping pool", "error", err)
-		}
-
 		pool.Close()
 		return nil, err
 	}

internal/repository/check/execution.go

@@ -0,0 +1,72 @@
+package check
+
+import (
+	"context"
+	"encoding/json"
+
+	e2 "github.com/pixel365/pulse/internal/e"
+	"github.com/pixel365/pulse/internal/model"
+	"github.com/pixel365/pulse/internal/repository"
+)
+
+var _ CheckExecutionRepository = (*ExecutionCheck)(nil)
+
+type ExecutionCheck struct {
+	db repository.QueryExecutor
+}
+
+func (e *ExecutionCheck) Add(ctx context.Context, result model.CheckExecutionResult) error {
+	query := `
+INSERT INTO pulse.check_executions (
+    execution_id,
+    check_id,
+    service_id,
+    status,
+    check_type,
+    started_at,
+    finished_at,
+    duration,
+    attempts_total,
+    error_kind,
+    error_message,
+    details
+) VALUES (
+    $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12
+)
+`
+
+	var details []byte
+	if result.Details != nil {
+		data, err := json.Marshal(result.Details)
+		if err != nil {
+			return err
+		}
+		details = data
+	}
+
+	var errKind string
+	if result.ErrorKind != e2.ErrNone {
+		errKind = string(result.ErrorKind)
+	}
+
+	_, err := e.db.Exec(ctx, query,
+		result.ExecutionID,
+		result.CheckID,
+		result.ServiceID,
+		result.Status,
+		result.CheckType,
+		result.StartedAt,
+		result.FinishedAt,
+		result.Duration,
+		result.AttemptsTotal,
+		&errKind,
+		result.ErrorMessage,
+		details,
+	)
+
+	return err
+}
+
+func NewExecutionRepository(db repository.QueryExecutor) *ExecutionCheck {
+	return &ExecutionCheck{db}
+}

internal/repository/check/interfaces.go

@@ -10,3 +10,7 @@ type CheckStateRepository interface {
 	GetCheckState(context.Context, string) (*model.CheckState, error)
 	UpdateCheckState(context.Context, model.CheckState) error
 }
+
+type CheckExecutionRepository interface {
+	Add(context.Context, model.CheckExecutionResult) error
+}

internal/repository/check/state.go

@@ -4,11 +4,13 @@ import (
 	"context"
 
 	"github.com/pixel365/pulse/internal/model"
+	"github.com/pixel365/pulse/internal/repository"
 )
 
 var _ CheckStateRepository = (*StateCheck)(nil)
 
 type StateCheck struct {
+	db repository.QueryExecutor
 }
 
 func (s *StateCheck) GetCheckState(ctx context.Context, service string) (*model.CheckState, error) {
@@ -19,6 +21,6 @@ func (s *StateCheck) UpdateCheckState(ctx context.Context, state model.CheckStat
 	return nil
 }
 
-func NewStateRepository() *StateCheck {
-	return &StateCheck{}
+func NewStateRepository(db repository.QueryExecutor) *StateCheck {
+	return &StateCheck{db}
 }

internal/repository/db.go

@@ -0,0 +1,16 @@
+package repository
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+)
+
+type QueryExecutor interface {
+	Exec(context.Context, string, ...any) (pgconn.CommandTag, error)
+	Query(context.Context, string, ...any) (pgx.Rows, error)
+	QueryRow(context.Context, string, ...any) pgx.Row
+	BeginTx(context.Context, pgx.TxOptions) (pgx.Tx, error)
+	Close()
+}

internal/services/check/handler.go

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