feat: telemetry alert rules with webhook delivery

core.alert_rules (migration 00012) — heartbeat_silence and error_spike
rules evaluated by a 60s backend ticker. Transition-based firing:
alert.fired on breach, alert.resolved on recovery, both first-class
webhook event types riding the signed dispatcher + realtime stream.
Dashboard manages rules on the Integrations page.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
Tomas DvorakandDevin committed 2026-09-24 19:09:19 +02:00
1 parent 345cf7918b
commit 2f4de543a9
28 files changed
+1693 -10

No files matched your search

@@ -0,0 +1,16 @@
-- +goose Up
CREATE TABLE core.alert_rules (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id UUID NOT NULL REFERENCES core.projects(id) ON DELETE CASCADE,
name TEXT NOT NULL,
kind TEXT NOT NULL CHECK (kind IN ('heartbeat_silence','error_spike')),
config JSONB NOT NULL DEFAULT '{}'::jsonb,
enabled BOOLEAN NOT NULL DEFAULT TRUE,
state JSONB NOT NULL DEFAULT '{}'::jsonb,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE (project_id, name)
);
-- +goose Down
DROP TABLE IF EXISTS core.alert_rules;
+49
View File
@@ -0,0 +1,49 @@
-- name: CreateAlertRule :one
INSERT INTO core.alert_rules (project_id, name, kind, config)
VALUES ($1, $2, $3, $4)
RETURNING *;
-- name: ListAlertRules :many
SELECT * FROM core.alert_rules
WHERE project_id = $1
ORDER BY name;
-- name: ListEnabledAlertRules :many
SELECT * FROM core.alert_rules
WHERE enabled;
-- name: GetAlertRule :one
SELECT * FROM core.alert_rules
WHERE id = $1 AND project_id = $2;
-- name: UpdateAlertRule :one
UPDATE core.alert_rules
SET name = $3, config = $4, enabled = $5, updated_at = NOW()
WHERE id = $1 AND project_id = $2
RETURNING *;
-- name: SetAlertRuleState :exec
UPDATE core.alert_rules
SET state = $3, updated_at = NOW()
WHERE id = $1 AND project_id = $2;
-- name: DeleteAlertRule :exec
DELETE FROM core.alert_rules
WHERE id = $1 AND project_id = $2;
-- Per-component most recent heartbeat — heartbeat_silence detection.
-- name: LastHeartbeats :many
SELECT co.name AS component_name, max(e.ts)::timestamptz AS last_beat
FROM core.components co
JOIN core.events e ON e.component_id = co.id AND e.type = 'heartbeat'
WHERE co.project_id = $1
GROUP BY co.name;
-- Per-component error count inside a sliding window — error_spike detection.
-- name: RecentErrorCounts :many
SELECT co.name AS component_name, count(*) AS errors
FROM core.events e
JOIN core.components co ON co.id = e.component_id
WHERE e.project_id = $1 AND e.type = 'error'
AND e.ts > now() - make_interval(secs => $2::bigint)
GROUP BY co.name;
@@ -0,0 +1,290 @@
// Code generated by sqlc. DO NOT EDIT.
// versions:
// sqlc v1.29.0
// source: alert_rules.sql
package db
import (
"context"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
)
const createAlertRule = `-- name: CreateAlertRule :one
INSERT INTO core.alert_rules (project_id, name, kind, config)
VALUES ($1, $2, $3, $4)
RETURNING id, project_id, name, kind, config, enabled, state, created_at, updated_at
`
type CreateAlertRuleParams struct {
ProjectID uuid.UUID `json:"project_id"`
Name string `json:"name"`
Kind string `json:"kind"`
Config []byte `json:"config"`
}
func (q *Queries) CreateAlertRule(ctx context.Context, arg CreateAlertRuleParams) (CoreAlertRule, error) {
row := q.db.QueryRow(ctx, createAlertRule,
arg.ProjectID,
arg.Name,
arg.Kind,
arg.Config,
)
var i CoreAlertRule
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.Kind,
&i.Config,
&i.Enabled,
&i.State,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
const deleteAlertRule = `-- name: DeleteAlertRule :exec
DELETE FROM core.alert_rules
WHERE id = $1 AND project_id = $2
`
type DeleteAlertRuleParams struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
}
func (q *Queries) DeleteAlertRule(ctx context.Context, arg DeleteAlertRuleParams) error {
_, err := q.db.Exec(ctx, deleteAlertRule, arg.ID, arg.ProjectID)
return err
}
const getAlertRule = `-- name: GetAlertRule :one
SELECT id, project_id, name, kind, config, enabled, state, created_at, updated_at FROM core.alert_rules
WHERE id = $1 AND project_id = $2
`
type GetAlertRuleParams struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
}
func (q *Queries) GetAlertRule(ctx context.Context, arg GetAlertRuleParams) (CoreAlertRule, error) {
row := q.db.QueryRow(ctx, getAlertRule, arg.ID, arg.ProjectID)
var i CoreAlertRule
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.Kind,
&i.Config,
&i.Enabled,
&i.State,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
const lastHeartbeats = `-- name: LastHeartbeats :many
SELECT co.name AS component_name, max(e.ts)::timestamptz AS last_beat
FROM core.components co
JOIN core.events e ON e.component_id = co.id AND e.type = 'heartbeat'
WHERE co.project_id = $1
GROUP BY co.name
`
type LastHeartbeatsRow struct {
ComponentName string `json:"component_name"`
LastBeat pgtype.Timestamptz `json:"last_beat"`
}
// Per-component most recent heartbeat — heartbeat_silence detection.
func (q *Queries) LastHeartbeats(ctx context.Context, projectID uuid.UUID) ([]LastHeartbeatsRow, error) {
rows, err := q.db.Query(ctx, lastHeartbeats, projectID)
if err != nil {
return nil, err
}
defer rows.Close()
var items []LastHeartbeatsRow
for rows.Next() {
var i LastHeartbeatsRow
if err := rows.Scan(&i.ComponentName, &i.LastBeat); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listAlertRules = `-- name: ListAlertRules :many
SELECT id, project_id, name, kind, config, enabled, state, created_at, updated_at FROM core.alert_rules
WHERE project_id = $1
ORDER BY name
`
func (q *Queries) ListAlertRules(ctx context.Context, projectID uuid.UUID) ([]CoreAlertRule, error) {
rows, err := q.db.Query(ctx, listAlertRules, projectID)
if err != nil {
return nil, err
}
defer rows.Close()
var items []CoreAlertRule
for rows.Next() {
var i CoreAlertRule
if err := rows.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.Kind,
&i.Config,
&i.Enabled,
&i.State,
&i.CreatedAt,
&i.UpdatedAt,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const listEnabledAlertRules = `-- name: ListEnabledAlertRules :many
SELECT id, project_id, name, kind, config, enabled, state, created_at, updated_at FROM core.alert_rules
WHERE enabled
`
func (q *Queries) ListEnabledAlertRules(ctx context.Context) ([]CoreAlertRule, error) {
rows, err := q.db.Query(ctx, listEnabledAlertRules)
if err != nil {
return nil, err
}
defer rows.Close()
var items []CoreAlertRule
for rows.Next() {
var i CoreAlertRule
if err := rows.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.Kind,
&i.Config,
&i.Enabled,
&i.State,
&i.CreatedAt,
&i.UpdatedAt,
); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const recentErrorCounts = `-- name: RecentErrorCounts :many
SELECT co.name AS component_name, count(*) AS errors
FROM core.events e
JOIN core.components co ON co.id = e.component_id
WHERE e.project_id = $1 AND e.type = 'error'
AND e.ts > now() - make_interval(secs => $2::bigint)
GROUP BY co.name
`
type RecentErrorCountsParams struct {
ProjectID uuid.UUID `json:"project_id"`
Column2 int64 `json:"column_2"`
}
type RecentErrorCountsRow struct {
ComponentName string `json:"component_name"`
Errors int64 `json:"errors"`
}
// Per-component error count inside a sliding window — error_spike detection.
func (q *Queries) RecentErrorCounts(ctx context.Context, arg RecentErrorCountsParams) ([]RecentErrorCountsRow, error) {
rows, err := q.db.Query(ctx, recentErrorCounts, arg.ProjectID, arg.Column2)
if err != nil {
return nil, err
}
defer rows.Close()
var items []RecentErrorCountsRow
for rows.Next() {
var i RecentErrorCountsRow
if err := rows.Scan(&i.ComponentName, &i.Errors); err != nil {
return nil, err
}
items = append(items, i)
}
if err := rows.Err(); err != nil {
return nil, err
}
return items, nil
}
const setAlertRuleState = `-- name: SetAlertRuleState :exec
UPDATE core.alert_rules
SET state = $3, updated_at = NOW()
WHERE id = $1 AND project_id = $2
`
type SetAlertRuleStateParams struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
State []byte `json:"state"`
}
func (q *Queries) SetAlertRuleState(ctx context.Context, arg SetAlertRuleStateParams) error {
_, err := q.db.Exec(ctx, setAlertRuleState, arg.ID, arg.ProjectID, arg.State)
return err
}
const updateAlertRule = `-- name: UpdateAlertRule :one
UPDATE core.alert_rules
SET name = $3, config = $4, enabled = $5, updated_at = NOW()
WHERE id = $1 AND project_id = $2
RETURNING id, project_id, name, kind, config, enabled, state, created_at, updated_at
`
type UpdateAlertRuleParams struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
Name string `json:"name"`
Config []byte `json:"config"`
Enabled bool `json:"enabled"`
}
func (q *Queries) UpdateAlertRule(ctx context.Context, arg UpdateAlertRuleParams) (CoreAlertRule, error) {
row := q.db.QueryRow(ctx, updateAlertRule,
arg.ID,
arg.ProjectID,
arg.Name,
arg.Config,
arg.Enabled,
)
var i CoreAlertRule
err := row.Scan(
&i.ID,
&i.ProjectID,
&i.Name,
&i.Kind,
&i.Config,
&i.Enabled,
&i.State,
&i.CreatedAt,
&i.UpdatedAt,
)
return i, err
}
@@ -140,6 +140,18 @@ func (ns NullCoreProjectRole) Value() (driver.Value, error) {
return string(ns.CoreProjectRole), nil
}
type CoreAlertRule struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
Name string `json:"name"`
Kind string `json:"kind"`
Config []byte `json:"config"`
Enabled bool `json:"enabled"`
State []byte `json:"state"`
CreatedAt pgtype.Timestamptz `json:"created_at"`
UpdatedAt pgtype.Timestamptz `json:"updated_at"`
}
type CoreApiKey struct {
ID uuid.UUID `json:"id"`
ProjectID uuid.UUID `json:"project_id"`
@@ -23,6 +23,7 @@ type Querier interface {
CountOrganizations(ctx context.Context) (int64, error)
CountProjectAdmins(ctx context.Context, projectID uuid.UUID) (int64, error)
CreateAPIKey(ctx context.Context, arg CreateAPIKeyParams) (CoreApiKey, error)
CreateAlertRule(ctx context.Context, arg CreateAlertRuleParams) (CoreAlertRule, error)
CreateAuditLog(ctx context.Context, arg CreateAuditLogParams) (CoreAuditLog, error)
CreateBucket(ctx context.Context, arg CreateBucketParams) (CoreBucket, error)
CreateBucketObject(ctx context.Context, arg CreateBucketObjectParams) (CoreBucketObject, error)
@@ -35,6 +36,7 @@ type Querier interface {
CreateProject(ctx context.Context, arg CreateProjectParams) (CoreProject, error)
CreateScheduledJob(ctx context.Context, arg CreateScheduledJobParams) (CoreScheduledJob, error)
CreateWebhook(ctx context.Context, arg CreateWebhookParams) (CoreWebhook, error)
DeleteAlertRule(ctx context.Context, arg DeleteAlertRuleParams) error
DeleteBucketByID(ctx context.Context, id uuid.UUID) (CoreBucket, error)
DeleteBucketObjectByKey(ctx context.Context, arg DeleteBucketObjectByKeyParams) (CoreBucketObject, error)
DeleteCollection(ctx context.Context, arg DeleteCollectionParams) error
@@ -54,6 +56,7 @@ type Querier interface {
FinishScheduledJobRun(ctx context.Context, arg FinishScheduledJobRunParams) (CoreScheduledJobRun, error)
GetAPIKeyByIDForProject(ctx context.Context, arg GetAPIKeyByIDForProjectParams) (CoreApiKey, error)
GetAPIKeyByPrefix(ctx context.Context, prefix string) (GetAPIKeyByPrefixRow, error)
GetAlertRule(ctx context.Context, arg GetAlertRuleParams) (CoreAlertRule, error)
GetBucketByID(ctx context.Context, id uuid.UUID) (GetBucketByIDRow, error)
GetBucketObjectByKey(ctx context.Context, arg GetBucketObjectByKeyParams) (CoreBucketObject, error)
GetCollectionByID(ctx context.Context, id uuid.UUID) (CoreCollection, error)
@@ -80,7 +83,10 @@ type Querier interface {
InsertEvent(ctx context.Context, arg InsertEventParams) (CoreEvent, error)
InsertScheduledJobRun(ctx context.Context, arg InsertScheduledJobRunParams) (CoreScheduledJobRun, error)
InsertWebhookDelivery(ctx context.Context, arg InsertWebhookDeliveryParams) (CoreWebhookDelivery, error)
// Per-component most recent heartbeat — heartbeat_silence detection.
LastHeartbeats(ctx context.Context, projectID uuid.UUID) ([]LastHeartbeatsRow, error)
ListAPIKeysForProject(ctx context.Context, projectID uuid.UUID) ([]CoreApiKey, error)
ListAlertRules(ctx context.Context, projectID uuid.UUID) ([]CoreAlertRule, error)
ListAuditLogsForProject(ctx context.Context, arg ListAuditLogsForProjectParams) ([]CoreAuditLog, error)
ListBucketObjects(ctx context.Context, arg ListBucketObjectsParams) ([]CoreBucketObject, error)
ListBucketsForOrganization(ctx context.Context, organizationID uuid.UUID) ([]uuid.UUID, error)
@@ -90,6 +96,7 @@ type Querier interface {
ListDBConnections(ctx context.Context, projectID uuid.UUID) ([]CoreDbConnection, error)
ListDocuments(ctx context.Context, arg ListDocumentsParams) ([]CoreDocument, error)
ListDueScheduledJobs(ctx context.Context, nextRunAt pgtype.Timestamptz) ([]CoreScheduledJob, error)
ListEnabledAlertRules(ctx context.Context) ([]CoreAlertRule, error)
ListErrorGroups(ctx context.Context, arg ListErrorGroupsParams) ([]ListErrorGroupsRow, error)
ListEvents(ctx context.Context, arg ListEventsParams) ([]ListEventsRow, error)
ListIntegrations(ctx context.Context, projectID uuid.UUID) ([]CoreIntegration, error)
@@ -114,10 +121,13 @@ type Querier interface {
MetricSeries(ctx context.Context, arg MetricSeriesParams) ([]MetricSeriesRow, error)
MoveBucketObject(ctx context.Context, arg MoveBucketObjectParams) (CoreBucketObject, error)
PruneScheduledJobRuns(ctx context.Context, arg PruneScheduledJobRunsParams) error
// Per-component error count inside a sliding window — error_spike detection.
RecentErrorCounts(ctx context.Context, arg RecentErrorCountsParams) ([]RecentErrorCountsRow, error)
RemoveOrganizationMember(ctx context.Context, arg RemoveOrganizationMemberParams) (CoreOrganizationMember, error)
RemoveProjectMember(ctx context.Context, arg RemoveProjectMemberParams) (CoreProjectMember, error)
RemoveProjectMembershipsForOrganizationUser(ctx context.Context, arg RemoveProjectMembershipsForOrganizationUserParams) error
RevokeAPIKey(ctx context.Context, arg RevokeAPIKeyParams) (CoreApiKey, error)
SetAlertRuleState(ctx context.Context, arg SetAlertRuleStateParams) error
SweepExpiredAuditLogs(ctx context.Context) (int64, error)
// Per-project retention sweeps. A retention of 0 days disables the sweep for
// that stream. Pending webhook deliveries are never swept — they may still be
@@ -128,6 +138,7 @@ type Querier interface {
SweepExpiredOrgAuditLogs(ctx context.Context) (int64, error)
SweepExpiredWebhookDeliveries(ctx context.Context) (int64, error)
TouchAPIKey(ctx context.Context, id uuid.UUID) error
UpdateAlertRule(ctx context.Context, arg UpdateAlertRuleParams) (CoreAlertRule, error)
UpdateBucketByID(ctx context.Context, arg UpdateBucketByIDParams) (CoreBucket, error)
UpdateCollection(ctx context.Context, arg UpdateCollectionParams) (CoreCollection, error)
UpdateDocument(ctx context.Context, arg UpdateDocumentParams) (CoreDocument, error)
+97
View File
@@ -0,0 +1,97 @@
package handlers
import (
"errors"
"net/http"
"strings"
"github.com/gin-gonic/gin"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/tdvorak/primora/apps/backend/internal/middleware"
apperrors "github.com/tdvorak/primora/apps/backend/internal/response"
"github.com/tdvorak/primora/apps/backend/internal/services"
)
func (h *HTTPHandler) alertError(c *gin.Context, err error) {
if errors.Is(err, pgx.ErrNoRows) {
apperrors.Abort(c, http.StatusNotFound, "not_found", "alert rule not found")
return
}
msg := err.Error()
if strings.Contains(msg, "invalid") || strings.Contains(msg, "must be") || strings.Contains(msg, "already exists") || strings.Contains(msg, "too long") {
apperrors.Abort(c, http.StatusBadRequest, "bad_request", msg)
return
}
h.handleError(c, err)
}
func (h *HTTPHandler) listAlertRules(c *gin.Context) {
actor, projectID, ok := h.actorAndProject(c)
if !ok {
return
}
result, err := h.Platform.ListAlertRules(c.Request.Context(), actor, projectID)
if err != nil {
h.alertError(c, err)
return
}
c.JSON(http.StatusOK, gin.H{"items": result})
}
func (h *HTTPHandler) createAlertRule(c *gin.Context) {
actor, projectID, ok := h.actorAndProject(c)
if !ok {
return
}
var body services.UpsertAlertRuleInput
if !h.bindAndValidate(c, &body) {
return
}
result, err := h.Platform.CreateAlertRule(c.Request.Context(), actor, projectID, body, middleware.RequestIDFromContext(c))
if err != nil {
h.alertError(c, err)
return
}
c.JSON(http.StatusCreated, result)
}
func (h *HTTPHandler) updateAlertRule(c *gin.Context) {
actor, projectID, ok := h.actorAndProject(c)
if !ok {
return
}
ruleID, err := uuid.Parse(c.Param("ruleID"))
if err != nil {
apperrors.Abort(c, http.StatusBadRequest, "bad_request", "invalid alert rule id")
return
}
var body services.UpsertAlertRuleInput
if !h.bindAndValidate(c, &body) {
return
}
result, err := h.Platform.UpdateAlertRule(c.Request.Context(), actor, projectID, ruleID, body, middleware.RequestIDFromContext(c))
if err != nil {
h.alertError(c, err)
return
}
c.JSON(http.StatusOK, result)
}
func (h *HTTPHandler) deleteAlertRule(c *gin.Context) {
actor, projectID, ok := h.actorAndProject(c)
if !ok {
return
}
ruleID, err := uuid.Parse(c.Param("ruleID"))
if err != nil {
apperrors.Abort(c, http.StatusBadRequest, "bad_request", "invalid alert rule id")
return
}
if err := h.Platform.DeleteAlertRule(c.Request.Context(), actor, projectID, ruleID, middleware.RequestIDFromContext(c)); err != nil {
h.alertError(c, err)
return
}
c.Status(http.StatusNoContent)
}
+5
View File
@@ -121,6 +121,11 @@ func (h *HTTPHandler) Register(router *gin.Engine) {
api.PUT("/projects/:projectID/secrets/:name", h.setProjectSecret)
api.DELETE("/projects/:projectID/secrets/:name", h.deleteProjectSecret)
api.POST("/projects/:projectID/secrets/:name/reveal", h.revealProjectSecret)
api.GET("/projects/:projectID/alerts", h.listAlertRules)
api.POST("/projects/:projectID/alerts", h.createAlertRule)
api.PUT("/projects/:projectID/alerts/:ruleID", h.updateAlertRule)
api.DELETE("/projects/:projectID/alerts/:ruleID", h.deleteAlertRule)
api.GET("/projects/:projectID/realtime/stream", h.realtimeStream)
api.GET("/buckets/:bucketID/objects", h.listObjects)
api.POST("/buckets/:bucketID/objects", h.uploadObject)
+368
View File
@@ -0,0 +1,368 @@
package services
import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"strings"
"sync"
"time"
"github.com/google/uuid"
db "github.com/tdvorak/primora/apps/backend/internal/database/db"
"github.com/tdvorak/primora/apps/backend/internal/models"
)
// Alert rules detect conditions in telemetry (component heartbeat silence,
// error-rate spikes) and fire `alert.fired` / `alert.resolved` domain events.
// Delivery reuses the webhook dispatcher + realtime stream — the evaluator is
// the only new machinery. Rules fire on state transitions only, so a flaky
// component produces one fired + one resolved, not a storm.
const (
AlertEventFired = "alert.fired"
AlertEventResolved = "alert.resolved"
AlertKindHeartbeatSilence = "heartbeat_silence"
AlertKindErrorSpike = "error_spike"
)
// AlertRuleConfig is the JSONB config per rule kind.
//
// heartbeat_silence: component beats less often than ThresholdMinutes.
// error_spike: component logs more than ThresholdCount errors in
// WindowMinutes.
// Component empty = evaluate every component in the project.
type AlertRuleConfig struct {
Component string `json:"component,omitempty"`
ThresholdMinutes int `json:"threshold_minutes,omitempty"`
ThresholdCount int64 `json:"threshold_count,omitempty"`
WindowMinutes int `json:"window_minutes,omitempty"`
}
type AlertRuleSummary struct {
ID string `json:"id"`
ProjectID string `json:"project_id"`
Name string `json:"name"`
Kind string `json:"kind"`
Config AlertRuleConfig `json:"config"`
Enabled bool `json:"enabled"`
Firing []string `json:"firing"`
CreatedAt time.Time `json:"created_at"`
UpdatedAt time.Time `json:"updated_at"`
}
type UpsertAlertRuleInput struct {
Name string `json:"name" validate:"required"`
Kind string `json:"kind" validate:"required"`
Config AlertRuleConfig `json:"config"`
Enabled *bool `json:"enabled"`
}
func (s *PlatformService) validateAlertRule(input UpsertAlertRuleInput) (AlertRuleConfig, error) {
if !projectSecretNameRe.MatchString(input.Name) {
return AlertRuleConfig{}, fmt.Errorf("invalid alert rule name %q — use [A-Za-z_][A-Za-z0-9_]*", input.Name)
}
cfg := input.Config
switch input.Kind {
case AlertKindHeartbeatSilence:
if cfg.ThresholdMinutes < 1 || cfg.ThresholdMinutes > 10080 {
return cfg, errors.New("threshold_minutes must be 1–10080")
}
case AlertKindErrorSpike:
if cfg.ThresholdCount < 1 {
return cfg, errors.New("threshold_count must be ≥ 1")
}
if cfg.WindowMinutes == 0 {
cfg.WindowMinutes = 15
}
if cfg.WindowMinutes < 1 || cfg.WindowMinutes > 1440 {
return cfg, errors.New("window_minutes must be 1–1440")
}
default:
return cfg, fmt.Errorf("invalid kind %q — %s or %s", input.Kind, AlertKindHeartbeatSilence, AlertKindErrorSpike)
}
if cfg.Component != "" && len(cfg.Component) > 128 {
return cfg, errors.New("component name too long")
}
return cfg, nil
}
func alertRuleSummary(row db.CoreAlertRule) AlertRuleSummary {
var cfg AlertRuleConfig
_ = json.Unmarshal(row.Config, &cfg)
firing := parseFiring(row.State)
return AlertRuleSummary{
ID: row.ID.String(),
ProjectID: row.ProjectID.String(),
Name: row.Name,
Kind: row.Kind,
Config: cfg,
Enabled: row.Enabled,
Firing: firing.list(),
CreatedAt: row.CreatedAt.Time,
UpdatedAt: row.UpdatedAt.Time,
}
}
func (s *PlatformService) ListAlertRules(ctx context.Context, actor *models.Actor, projectID uuid.UUID) ([]AlertRuleSummary, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer", "viewer"); err != nil {
return nil, err
}
rows, err := s.repo.Queries().ListAlertRules(ctx, projectID)
if err != nil {
return nil, err
}
out := make([]AlertRuleSummary, len(rows))
for i, r := range rows {
out[i] = alertRuleSummary(r)
}
return out, nil
}
func (s *PlatformService) CreateAlertRule(ctx context.Context, actor *models.Actor, projectID uuid.UUID, input UpsertAlertRuleInput, requestID string) (AlertRuleSummary, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil {
return AlertRuleSummary{}, err
}
cfg, err := s.validateAlertRule(input)
if err != nil {
return AlertRuleSummary{}, err
}
rawCfg, _ := json.Marshal(cfg)
row, err := s.repo.Queries().CreateAlertRule(ctx, db.CreateAlertRuleParams{
ProjectID: projectID,
Name: input.Name,
Kind: input.Kind,
Config: rawCfg,
})
if err != nil {
if strings.Contains(err.Error(), "alert_rules_project_id_name_key") {
return AlertRuleSummary{}, fmt.Errorf("an alert rule named %s already exists", input.Name)
}
return AlertRuleSummary{}, err
}
s.auditAlert(ctx, actor, projectID, requestID, "alert.created", row.ID, input.Name)
return alertRuleSummary(row), nil
}
func (s *PlatformService) UpdateAlertRule(ctx context.Context, actor *models.Actor, projectID, ruleID uuid.UUID, input UpsertAlertRuleInput, requestID string) (AlertRuleSummary, error) {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil {
return AlertRuleSummary{}, err
}
if _, err := s.repo.Queries().GetAlertRule(ctx, db.GetAlertRuleParams{ID: ruleID, ProjectID: projectID}); err != nil {
return AlertRuleSummary{}, err // pgx.ErrNoRows → 404 at the handler
}
cfg, err := s.validateAlertRule(input)
if err != nil {
return AlertRuleSummary{}, err
}
enabled := true
if input.Enabled != nil {
enabled = *input.Enabled
}
rawCfg, _ := json.Marshal(cfg)
row, err := s.repo.Queries().UpdateAlertRule(ctx, db.UpdateAlertRuleParams{
ID: ruleID, ProjectID: projectID, Name: input.Name, Config: rawCfg, Enabled: enabled,
})
if err != nil {
return AlertRuleSummary{}, err
}
s.auditAlert(ctx, actor, projectID, requestID, "alert.updated", row.ID, input.Name)
return alertRuleSummary(row), nil
}
func (s *PlatformService) DeleteAlertRule(ctx context.Context, actor *models.Actor, projectID, ruleID uuid.UUID, requestID string) error {
if err := s.requireProjectRole(ctx, actor, projectID, "admin", "developer"); err != nil {
return err
}
if _, err := s.repo.Queries().GetAlertRule(ctx, db.GetAlertRuleParams{ID: ruleID, ProjectID: projectID}); err != nil {
return err
}
if err := s.repo.Queries().DeleteAlertRule(ctx, db.DeleteAlertRuleParams{ID: ruleID, ProjectID: projectID}); err != nil {
return err
}
s.auditAlert(ctx, actor, projectID, requestID, "alert.deleted", ruleID, "")
return nil
}
func (s *PlatformService) auditAlert(ctx context.Context, actor *models.Actor, projectID uuid.UUID, requestID, action string, ruleID uuid.UUID, name string) {
project, err := s.repo.Queries().GetProjectByID(ctx, projectID)
if err != nil {
return
}
_, _ = s.repo.Queries().CreateAuditLog(ctx, newAuditParams(project.OrganizationID, projectID, actor, requestID, action, "alert_rule", ruleID.String(), map[string]any{"name": name}))
}
// ---------------------------------------------------------------------------
// evaluator — one ticker per backend, transition-based firing
// ---------------------------------------------------------------------------
type firingSet map[string]bool
func parseFiring(raw []byte) firingSet {
var state struct {
Firing map[string]bool `json:"firing"`
}
if err := json.Unmarshal(raw, &state); err != nil || state.Firing == nil {
return firingSet{}
}
return state.Firing
}
func (f firingSet) list() []string {
out := make([]string, 0, len(f))
for name, on := range f {
if on {
out = append(out, name)
}
}
return out
}
// alertQueries is the slice of db.Queries the evaluator needs — narrow so
// tests can stub it without a database.
type alertQueries interface {
ListEnabledAlertRules(ctx context.Context) ([]db.CoreAlertRule, error)
LastHeartbeats(ctx context.Context, projectID uuid.UUID) ([]db.LastHeartbeatsRow, error)
RecentErrorCounts(ctx context.Context, arg db.RecentErrorCountsParams) ([]db.RecentErrorCountsRow, error)
SetAlertRuleState(ctx context.Context, arg db.SetAlertRuleStateParams) error
}
type alertEvaluator struct {
q alertQueries
publish func(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any)
logger *slog.Logger
tick time.Duration
quit chan struct{}
wg sync.WaitGroup
started sync.Once
}
func newAlertEvaluator(q alertQueries, publish func(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any), logger *slog.Logger) *alertEvaluator {
return &alertEvaluator{q: q, publish: publish, logger: logger, tick: time.Minute, quit: make(chan struct{})}
}
func (e *alertEvaluator) Start(ctx context.Context) {
e.started.Do(func() {
e.wg.Add(1)
go func() {
defer e.wg.Done()
t := time.NewTicker(e.tick)
defer t.Stop()
for {
select {
case <-t.C:
e.scan(ctx)
case <-e.quit:
return
case <-ctx.Done():
return
}
}
}()
})
}
func (e *alertEvaluator) Stop() {
close(e.quit)
e.wg.Wait()
}
func (e *alertEvaluator) scan(ctx context.Context) {
ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
rules, err := e.q.ListEnabledAlertRules(ctx)
if err != nil {
e.logger.Warn("alert scan: list rules", "error", err)
return
}
for _, rule := range rules {
if err := e.evaluate(ctx, rule); err != nil {
e.logger.Warn("alert scan: evaluate", "rule", rule.Name, "error", err)
}
}
}
func (e *alertEvaluator) evaluate(ctx context.Context, rule db.CoreAlertRule) error {
var cfg AlertRuleConfig
if err := json.Unmarshal(rule.Config, &cfg); err != nil {
return nil // bad config can't be fixed by re-scanning; skip quietly
}
now := time.Now()
var breached map[string]float64 // component → measured value
var metric, unit string
switch rule.Kind {
case AlertKindHeartbeatSilence:
rows, err := e.q.LastHeartbeats(ctx, rule.ProjectID)
if err != nil {
return err
}
breached = map[string]float64{}
for _, r := range rows {
silent := now.Sub(r.LastBeat.Time).Minutes()
if silent >= float64(cfg.ThresholdMinutes) {
breached[r.ComponentName] = silent
}
}
metric, unit = "silence", "minutes"
case AlertKindErrorSpike:
rows, err := e.q.RecentErrorCounts(ctx, db.RecentErrorCountsParams{
ProjectID: rule.ProjectID,
Column2: int64(cfg.WindowMinutes) * 60,
})
if err != nil {
return err
}
breached = map[string]float64{}
for _, r := range rows {
if r.Errors >= cfg.ThresholdCount {
breached[r.ComponentName] = float64(r.Errors)
}
}
metric, unit = "errors", "count"
default:
return nil
}
firing := parseFiring(rule.State)
changed := false
// newly breached components → fired
for comp, value := range breached {
if cfg.Component != "" && comp != cfg.Component {
continue
}
if firing[comp] {
continue
}
firing[comp] = true
changed = true
e.publish(ctx, rule.ProjectID, AlertEventFired, map[string]any{
"rule_id": rule.ID.String(), "rule": rule.Name, "kind": rule.Kind,
"component": comp, "metric": metric, "value": value, "unit": unit,
})
}
// recovered components → resolved
for comp, on := range firing {
_, stillBreached := breached[comp]
if !on || stillBreached {
continue
}
firing[comp] = false
changed = true
e.publish(ctx, rule.ProjectID, AlertEventResolved, map[string]any{
"rule_id": rule.ID.String(), "rule": rule.Name, "kind": rule.Kind,
"component": comp,
})
}
if !changed {
return nil
}
raw, _ := json.Marshal(map[string]any{"firing": firing})
return e.q.SetAlertRuleState(ctx, db.SetAlertRuleStateParams{ID: rule.ID, ProjectID: rule.ProjectID, State: raw})
}
@@ -0,0 +1,162 @@
package services
import (
"context"
"encoding/json"
"log/slog"
"testing"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5/pgtype"
db "github.com/tdvorak/primora/apps/backend/internal/database/db"
)
type stubAlertQueries struct {
rules []db.CoreAlertRule
beats []db.LastHeartbeatsRow
errors []db.RecentErrorCountsRow
stateSaved map[uuid.UUID][]byte
}
func (s *stubAlertQueries) ListEnabledAlertRules(ctx context.Context) ([]db.CoreAlertRule, error) {
return s.rules, nil
}
func (s *stubAlertQueries) LastHeartbeats(ctx context.Context, projectID uuid.UUID) ([]db.LastHeartbeatsRow, error) {
return s.beats, nil
}
func (s *stubAlertQueries) RecentErrorCounts(ctx context.Context, arg db.RecentErrorCountsParams) ([]db.RecentErrorCountsRow, error) {
return s.errors, nil
}
func (s *stubAlertQueries) SetAlertRuleState(ctx context.Context, arg db.SetAlertRuleStateParams) error {
if s.stateSaved == nil {
s.stateSaved = map[uuid.UUID][]byte{}
}
s.stateSaved[arg.ID] = arg.State
// persist back so subsequent evaluations see the transition
for i, r := range s.rules {
if r.ID == arg.ID {
s.rules[i].State = arg.State
}
}
return nil
}
func rule(t *testing.T, kind string, cfg AlertRuleConfig) db.CoreAlertRule {
raw, err := json.Marshal(cfg)
if err != nil {
t.Fatal(err)
}
return db.CoreAlertRule{ID: uuid.New(), ProjectID: uuid.New(), Name: "r1", Kind: kind, Config: raw, State: []byte(`{"firing":{}}`)}
}
func collect() (func(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any), *[]string) {
events := &[]string{}
return func(ctx context.Context, projectID uuid.UUID, eventType string, data map[string]any) {
*events = append(*events, eventType+":"+data["component"].(string))
}, events
}
func TestAlertSilenceFiresOnceThenResolves(t *testing.T) {
r := rule(t, AlertKindHeartbeatSilence, AlertRuleConfig{ThresholdMinutes: 5})
stub := &stubAlertQueries{
rules: []db.CoreAlertRule{r},
beats: []db.LastHeartbeatsRow{{
ComponentName: "web",
LastBeat: pgtype.Timestamptz{Time: time.Now().Add(-10 * time.Minute), Valid: true},
}},
}
publish, events := collect()
e := &alertEvaluator{q: stub, publish: publish, logger: slog.Default()}
if err := e.evaluate(context.Background(), stub.rules[0]); err != nil {
t.Fatal(err)
}
if len(*events) != 1 || (*events)[0] != "alert.fired:web" {
t.Fatalf("want one alert.fired:web, got %v", *events)
}
// second evaluation — still silent, must not refire
if err := e.evaluate(context.Background(), stub.rules[0]); err != nil {
t.Fatal(err)
}
if len(*events) != 1 {
t.Fatalf("refired: %v", *events)
}
// heartbeat resumes → resolved
stub.beats[0].LastBeat = pgtype.Timestamptz{Time: time.Now(), Valid: true}
if err := e.evaluate(context.Background(), stub.rules[0]); err != nil {
t.Fatal(err)
}
if len(*events) != 2 || (*events)[1] != "alert.resolved:web" {
t.Fatalf("want alert.resolved:web, got %v", *events)
}
}
func TestAlertErrorSpikeThreshold(t *testing.T) {
r := rule(t, AlertKindErrorSpike, AlertRuleConfig{ThresholdCount: 3, WindowMinutes: 15})
stub := &stubAlertQueries{
rules: []db.CoreAlertRule{r},
errors: []db.RecentErrorCountsRow{{ComponentName: "api", Errors: 2}},
}
publish, events := collect()
e := &alertEvaluator{q: stub, publish: publish, logger: slog.Default()}
if err := e.evaluate(context.Background(), stub.rules[0]); err != nil {
t.Fatal(err)
}
if len(*events) != 0 {
t.Fatalf("under threshold fired: %v", *events)
}
stub.errors[0].Errors = 5
if err := e.evaluate(context.Background(), stub.rules[0]); err != nil {
t.Fatal(err)
}
if len(*events) != 1 || (*events)[0] != "alert.fired:api" {
t.Fatalf("want alert.fired:api, got %v", *events)
}
}
func TestAlertComponentFilter(t *testing.T) {
r := rule(t, AlertKindHeartbeatSilence, AlertRuleConfig{Component: "web", ThresholdMinutes: 5})
stub := &stubAlertQueries{
rules: []db.CoreAlertRule{r},
beats: []db.LastHeartbeatsRow{
{ComponentName: "web", LastBeat: pgtype.Timestamptz{Time: time.Now().Add(-time.Hour), Valid: true}},
{ComponentName: "api", LastBeat: pgtype.Timestamptz{Time: time.Now().Add(-time.Hour), Valid: true}},
},
}
publish, events := collect()
e := &alertEvaluator{q: stub, publish: publish, logger: slog.Default()}
if err := e.evaluate(context.Background(), stub.rules[0]); err != nil {
t.Fatal(err)
}
if len(*events) != 1 || (*events)[0] != "alert.fired:web" {
t.Fatalf("filtered rule must fire only for web, got %v", *events)
}
}
func TestValidateAlertRule(t *testing.T) {
s := &PlatformService{}
if _, err := s.validateAlertRule(UpsertAlertRuleInput{Name: "ok", Kind: AlertKindHeartbeatSilence, Config: AlertRuleConfig{ThresholdMinutes: 5}}); err != nil {
t.Fatalf("valid silence rule rejected: %v", err)
}
cfg, err := s.validateAlertRule(UpsertAlertRuleInput{Name: "ok", Kind: AlertKindErrorSpike, Config: AlertRuleConfig{ThresholdCount: 3}})
if err != nil || cfg.WindowMinutes != 15 {
t.Fatalf("window default: cfg=%+v err=%v", cfg, err)
}
for _, bad := range []UpsertAlertRuleInput{
{Name: "bad name!", Kind: AlertKindHeartbeatSilence, Config: AlertRuleConfig{ThresholdMinutes: 5}},
{Name: "x", Kind: "bogus", Config: AlertRuleConfig{}},
{Name: "x", Kind: AlertKindHeartbeatSilence, Config: AlertRuleConfig{ThresholdMinutes: 0}},
{Name: "x", Kind: AlertKindErrorSpike, Config: AlertRuleConfig{ThresholdCount: 0}},
} {
if _, err := s.validateAlertRule(bad); err == nil {
t.Fatalf("accepted invalid rule %+v", bad)
}
}
}
@@ -38,6 +38,7 @@ type PlatformService struct {
enc *secrets.Encryptor
dispatcher *WebhookDispatcher
scheduler *JobScheduler
alerts *alertEvaluator
}
type BootstrapInput struct {
@@ -230,6 +231,8 @@ func NewPlatformService(repo *repositories.CoreRepository, store *storage.LocalS
s.dispatcher.Start(context.Background())
s.scheduler = NewJobScheduler(repo.Queries(), enc, logger, s.publishEvent)
s.scheduler.Start(context.Background())
s.alerts = newAlertEvaluator(repo.Queries(), s.publishEvent, logger)
s.alerts.Start(context.Background())
}
return s
}
@@ -242,6 +245,9 @@ func (s *PlatformService) Close() {
if s.scheduler != nil {
s.scheduler.Stop()
}
if s.alerts != nil {
s.alerts.Stop()
}
}
// publishEvent fans a domain event out to realtime SSE subscribers and to
@@ -31,6 +31,8 @@ const (
WebhookEventIssueCreated = "issue.created"
WebhookEventDeployMarker = "deploy.marker"
WebhookEventTest = "webhook.test"
WebhookEventAlertFired = "alert.fired"
WebhookEventAlertResolved = "alert.resolved"
WebhookEventDocumentCreated = "document.created"
WebhookEventDocumentUpdated = "document.updated"
WebhookEventDocumentDeleted = "document.deleted"
@@ -43,6 +45,8 @@ var webhookEventTypes = map[string]bool{
WebhookEventIssueCreated: true,
WebhookEventDeployMarker: true,
WebhookEventTest: true,
WebhookEventAlertFired: true,
WebhookEventAlertResolved: true,
WebhookEventJobRun: true,
WebhookEventDocumentCreated: true,
WebhookEventDocumentUpdated: true,
+171 -3
View File
@@ -1941,6 +1941,106 @@ paths:
$ref: "#/components/responses/ErrorResponse"
"404":
$ref: "#/components/responses/ErrorResponse"
/projects/{projectID}/alerts:
get:
tags: [Alerts]
operationId: listAlertRules
summary: Lists telemetry alert rules for the project.
security:
- bearerAuth: []
- apiKeyAuth: []
parameters:
- $ref: "#/components/parameters/ProjectID"
responses:
"200":
description: Alert rules
content:
application/json:
schema:
$ref: "#/components/schemas/AlertRuleListResponse"
"401":
$ref: "#/components/responses/ErrorResponse"
post:
tags: [Alerts]
operationId: createAlertRule
summary: Creates an alert rule. Rules fire `alert.fired`/`alert.resolved` domain events on state transitions — subscribe a webhook to those event types to receive them.
security:
- bearerAuth: []
- apiKeyAuth: []
parameters:
- $ref: "#/components/parameters/ProjectID"
requestBody:
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/UpsertAlertRuleRequest"
responses:
"201":
description: Alert rule created
content:
application/json:
schema:
$ref: "#/components/schemas/AlertRule"
"400":
$ref: "#/components/responses/ErrorResponse"
"401":
$ref: "#/components/responses/ErrorResponse"
/projects/{projectID}/alerts/{ruleID}:
put:
tags: [Alerts]
operationId: updateAlertRule
security:
- bearerAuth: []
- apiKeyAuth: []
parameters:
- $ref: "#/components/parameters/ProjectID"
- name: ruleID
in: path
required: true
schema:
type: string
format: uuid
requestBody:
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/UpsertAlertRuleRequest"
responses:
"200":
description: Alert rule updated
content:
application/json:
schema:
$ref: "#/components/schemas/AlertRule"
"400":
$ref: "#/components/responses/ErrorResponse"
"401":
$ref: "#/components/responses/ErrorResponse"
"404":
$ref: "#/components/responses/ErrorResponse"
delete:
tags: [Alerts]
operationId: deleteAlertRule
security:
- bearerAuth: []
- apiKeyAuth: []
parameters:
- $ref: "#/components/parameters/ProjectID"
- name: ruleID
in: path
required: true
schema:
type: string
format: uuid
responses:
"204":
description: Alert rule deleted
"401":
$ref: "#/components/responses/ErrorResponse"
"404":
$ref: "#/components/responses/ErrorResponse"
components:
securitySchemes:
bearerAuth:
@@ -3512,7 +3612,7 @@ components:
type: array
items:
type: string
description: Event filter (issue.created, deploy.marker, webhook.test, job.run, document.*, object.*). Empty means all events.
description: Event filter (issue.created, deploy.marker, webhook.test, job.run, alert.fired, alert.resolved, document.*, object.*). Empty means all events.
enabled:
type: boolean
has_secret:
@@ -3534,7 +3634,7 @@ components:
type: array
items:
type: string
enum: [issue.created, deploy.marker, webhook.test, job.run, document.created, document.updated, document.deleted, object.created, object.updated, object.deleted]
enum: [issue.created, deploy.marker, webhook.test, job.run, alert.fired, alert.resolved, document.created, document.updated, document.deleted, object.created, object.updated, object.deleted]
enabled:
type: boolean
required: [url]
@@ -3551,7 +3651,7 @@ components:
type: array
items:
type: string
enum: [issue.created, deploy.marker, webhook.test, job.run, document.created, document.updated, document.deleted, object.created, object.updated, object.deleted]
enum: [issue.created, deploy.marker, webhook.test, job.run, alert.fired, alert.resolved, document.created, document.updated, document.deleted, object.created, object.updated, object.deleted]
enabled:
type: boolean
WebhookCreateResponse:
@@ -3821,3 +3921,71 @@ components:
value:
type: string
required: [name, value]
AlertRuleConfig:
type: object
properties:
component:
type: string
description: Restrict the rule to one component name; empty = all components.
threshold_minutes:
type: integer
description: heartbeat_silence — fire when a component has not heartbeated for this many minutes (1–10080).
threshold_count:
type: integer
format: int64
description: error_spike — fire when a component logs at least this many errors inside the window.
window_minutes:
type: integer
description: error_spike — sliding window size (default 15, 1–1440).
AlertRule:
type: object
properties:
id:
type: string
format: uuid
project_id:
type: string
format: uuid
name:
type: string
kind:
type: string
enum: [heartbeat_silence, error_spike]
config:
$ref: "#/components/schemas/AlertRuleConfig"
enabled:
type: boolean
firing:
type: array
items:
type: string
description: Component names currently in breached state.
created_at:
type: string
format: date-time
updated_at:
type: string
format: date-time
required: [id, project_id, name, kind, config, enabled, firing, created_at, updated_at]
AlertRuleListResponse:
type: object
properties:
items:
type: array
items:
$ref: "#/components/schemas/AlertRule"
required: [items]
UpsertAlertRuleRequest:
type: object
properties:
name:
type: string
description: Env-var-safe name ([A-Za-z_][A-Za-z0-9_]*).
kind:
type: string
enum: [heartbeat_silence, error_spike]
config:
$ref: "#/components/schemas/AlertRuleConfig"
enabled:
type: boolean
required: [name, kind]