feat(query): add execution history filters and time-bucket aggregation

Author: pixel365

internal/model/result.go

@@ -11,6 +11,7 @@ import (
 type CheckExecutionStatus string
 type CheckStateStatus string
 type ServiceStateStatus string
+type CheckExecutionBucket string
 
 const (
 	CheckExecutionSuccess CheckExecutionStatus = "success"
@@ -24,6 +25,11 @@ const (
 	ServiceStateHealthy   ServiceStateStatus = "healthy"
 	ServiceStateUnhealthy ServiceStateStatus = "unhealthy"
 	ServiceStateDegraded  ServiceStateStatus = "degraded"
+
+	CheckExecutionBucketSecond CheckExecutionBucket = "second"
+	CheckExecutionBucketMinute CheckExecutionBucket = "minute"
+	CheckExecutionBucketHour   CheckExecutionBucket = "hour"
+	CheckExecutionBucketDay    CheckExecutionBucket = "day"
 )
 
 type CheckExecutionResult struct {
@@ -81,7 +87,20 @@ type CheckExecutionFilter struct {
 	Limit     int
 }
 
-func (f *CheckExecutionFilter) Apply(query string, fieldFn func(string) string) (string, []any) {
+type CheckExecutionAggregateFilter struct {
+	Bucket CheckExecutionBucket
+	CheckExecutionFilter
+}
+
+type CheckExecutionBucketRecord struct {
+	BucketStart   time.Time
+	Total         int
+	SuccessCount  int
+	FailureCount  int
+	AvgDurationUs int64
+}
+
+func (f *CheckExecutionFilter) ApplyConditions(fieldFn func(string) string) ([]string, []any) {
 	var (
 		conditions []string
 		args       []any
@@ -99,14 +118,26 @@ func (f *CheckExecutionFilter) Apply(query string, fieldFn func(string) string)
 
 	if f.From != nil {
 		args = append(args, *f.From)
-		conditions = append(conditions, fmt.Sprintf("%s >= $%d", fieldFn("finished_at"), len(args)))
+		conditions = append(
+			conditions,
+			fmt.Sprintf("%s >= TIMESTAMPTZ $%d", fieldFn("finished_at"), len(args)),
+		)
 	}
 
 	if f.To != nil {
 		args = append(args, *f.To)
-		conditions = append(conditions, fmt.Sprintf("%s <= $%d", fieldFn("finished_at"), len(args)))
+		conditions = append(
+			conditions,
+			fmt.Sprintf("%s <= TIMESTAMPTZ $%d", fieldFn("finished_at"), len(args)),
+		)
 	}
 
+	return conditions, args
+}
+
+func (f *CheckExecutionFilter) Apply(query string, fieldFn func(string) string) (string, []any) {
+	conditions, args := f.ApplyConditions(fieldFn)
+
 	if len(conditions) > 0 {
 		query += "WHERE " + conditions[0]
 		for i := 1; i < len(conditions); i++ {

internal/repository/check/execution.go

@@ -3,6 +3,7 @@ package check
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"time"
 
 	"github.com/pixel365/pulse/internal/repository"
@@ -156,6 +157,69 @@ FROM pulse.check_executions
 	return result, nil
 }
 
+func (e *ExecutionCheck) ListExecutionBuckets(
+	ctx context.Context,
+	filter model.CheckExecutionAggregateFilter,
+) ([]model.CheckExecutionBucketRecord, error) {
+	bucketExpr, err := executionBucketExpr(filter.Bucket)
+	if err != nil {
+		return nil, err
+	}
+
+	query := fmt.Sprintf(`
+SELECT
+    %s AS bucket_start,
+    COUNT(1) AS total,
+    COUNT(1) FILTER (WHERE status = 'success') AS success_count,
+    COUNT(1) FILTER (WHERE status = 'failure') AS failure_count,
+    AVG(duration)::bigint AS avg_duration_us
+FROM pulse.check_executions
+`, bucketExpr)
+
+	conditions, args := filter.ApplyConditions(mustField)
+	if len(conditions) > 0 {
+		query += "WHERE " + conditions[0]
+		for i := 1; i < len(conditions); i++ {
+			query += " AND " + conditions[i]
+		}
+		query += "\n"
+	}
+
+	query += "GROUP BY bucket_start\n"
+	query += "ORDER BY bucket_start\n"
+
+	rows, err := e.db.Query(ctx, query, args...)
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var result []model.CheckExecutionBucketRecord
+
+	for rows.Next() {
+		var row model.CheckExecutionBucketRecord
+
+		err = rows.Scan(
+			&row.BucketStart,
+			&row.Total,
+			&row.SuccessCount,
+			&row.FailureCount,
+			&row.AvgDurationUs,
+		)
+		if err != nil {
+			return nil, err
+		}
+
+		result = append(result, row)
+	}
+
+	if err = rows.Err(); err != nil {
+		return nil, err
+	}
+
+	return result, nil
+}
+
 func mustField(name string) string {
 	switch name {
 	case "service_id", "check_id", "finished_at":
@@ -165,6 +229,21 @@ func mustField(name string) string {
 	panic("unknown field " + name + " in check execution filter")
 }
 
+func executionBucketExpr(bucket model.CheckExecutionBucket) (string, error) {
+	switch bucket {
+	case model.CheckExecutionBucketSecond:
+		return "date_trunc('second', finished_at)", nil
+	case "", model.CheckExecutionBucketMinute:
+		return "date_trunc('minute', finished_at)", nil
+	case model.CheckExecutionBucketHour:
+		return "date_trunc('hour', finished_at)", nil
+	case model.CheckExecutionBucketDay:
+		return "date_trunc('day', finished_at)", nil
+	default:
+		return "", fmt.Errorf("unsupported execution bucket %q", bucket)
+	}
+}
+
 func NewExecutionRepository(db repository.QueryExecutor) *ExecutionCheck {
 	return &ExecutionCheck{db}
 }

internal/repository/check/interfaces.go

@@ -18,4 +18,8 @@ type CheckExecutionRepository interface {
 		context.Context,
 		model.CheckExecutionFilter,
 	) ([]model.CheckExecutionRecord, error)
+	ListExecutionBuckets(
+		context.Context,
+		model.CheckExecutionAggregateFilter,
+	) ([]model.CheckExecutionBucketRecord, error)
 }