diff --git a/server/internal/services/monitors.go b/server/internal/services/monitors.go new file mode 100644 index 0000000..da00683 --- /dev/null +++ b/server/internal/services/monitors.go @@ -0,0 +1,241 @@ +package services + +import ( + "context" + "errors" + "time" + + "github.com/google/uuid" + "github.com/mrhid6/vantage/server/internal/checker" + "github.com/mrhid6/vantage/server/internal/db" + "github.com/mrhid6/vantage/server/internal/models" + "go.mongodb.org/mongo-driver/v2/bson" + "go.mongodb.org/mongo-driver/v2/mongo" + "go.mongodb.org/mongo-driver/v2/mongo/options" +) + +func monCtx() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), 5*time.Second) +} + +// SpecFor maps a monitor onto a checker.Spec. +func SpecFor(m *models.Monitor) checker.Spec { + return checker.Spec{ + Type: m.Type, + URL: m.Target.URL, + Host: m.Target.Host, + Port: m.Target.Port, + Method: m.Target.Method, + ExpectedStatus: m.Target.ExpectedStatus, + Keyword: m.Target.Keyword, + TLSWarnDays: m.Target.TLSWarnDays, + TimeoutSec: m.IntervalSec, + } +} + +func ListMonitors() ([]models.Monitor, error) { + ctx, cancel := monCtx() + defer cancel() + cur, err := db.Col("monitors").Find(ctx, bson.M{}, options.Find().SetSort(bson.M{"created_at": 1})) + if err != nil { + return nil, err + } + var out []models.Monitor + if err := cur.All(ctx, &out); err != nil { + return nil, err + } + return out, nil +} + +// ListMonitorsForRunner returns enabled monitors whose Runner matches runner. +func ListMonitorsForRunner(runner string) ([]models.Monitor, error) { + ctx, cancel := monCtx() + defer cancel() + cur, err := db.Col("monitors").Find(ctx, bson.M{"runner": runner, "enabled": true}) + if err != nil { + return nil, err + } + var out []models.Monitor + if err := cur.All(ctx, &out); err != nil { + return nil, err + } + return out, nil +} + +func GetMonitor(monitorID string) (*models.Monitor, error) { + ctx, cancel := monCtx() + defer cancel() + var m models.Monitor + err := db.Col("monitors").FindOne(ctx, bson.M{"monitor_id": monitorID}).Decode(&m) + if errors.Is(err, mongo.ErrNoDocuments) { + return nil, nil + } + if err != nil { + return nil, err + } + return &m, nil +} + +func CreateMonitor(m *models.Monitor) (*models.Monitor, error) { + ctx, cancel := monCtx() + defer cancel() + m.MonitorID = uuid.NewString() + m.CreatedAt = time.Now() + if m.IntervalSec <= 0 { + m.IntervalSec = 60 + } + if m.Retries <= 0 { + m.Retries = 1 + } + if m.Runner == "" { + m.Runner = models.RunnerServer + } + m.State = models.MonitorState{Status: models.StatusPending} + if _, err := db.Col("monitors").InsertOne(ctx, m); err != nil { + return nil, err + } + return m, nil +} + +func UpdateMonitor(monitorID string, upd bson.M) error { + ctx, cancel := monCtx() + defer cancel() + _, err := db.Col("monitors").UpdateOne(ctx, bson.M{"monitor_id": monitorID}, bson.M{"$set": upd}) + return err +} + +func DeleteMonitor(monitorID string) error { + ctx, cancel := monCtx() + defer cancel() + if _, err := db.Col("monitors").DeleteOne(ctx, bson.M{"monitor_id": monitorID}); err != nil { + return err + } + db.Col("incidents").DeleteMany(ctx, bson.M{"monitor_id": monitorID}) + db.Col("monitor_rollups").DeleteMany(ctx, bson.M{"monitor_id": monitorID}) + return nil +} + +func ListIncidents(monitorID string, limit int64) ([]models.Incident, error) { + ctx, cancel := monCtx() + defer cancel() + if limit <= 0 { + limit = 50 + } + cur, err := db.Col("incidents").Find(ctx, bson.M{"monitor_id": monitorID}, + options.Find().SetSort(bson.M{"started_at": -1}).SetLimit(limit)) + if err != nil { + return nil, err + } + var out []models.Incident + if err := cur.All(ctx, &out); err != nil { + return nil, err + } + return out, nil +} + +// UptimeRollups returns hourly rollups for a monitor since the cutoff, oldest first. +func UptimeRollups(monitorID string, since time.Time) ([]models.Rollup, error) { + ctx, cancel := monCtx() + defer cancel() + cur, err := db.Col("monitor_rollups").Find(ctx, + bson.M{"monitor_id": monitorID, "period_start": bson.M{"$gte": since}}, + options.Find().SetSort(bson.M{"period_start": 1})) + if err != nil { + return nil, err + } + var out []models.Rollup + if err := cur.All(ctx, &out); err != nil { + return nil, err + } + return out, nil +} + +// IngestResult applies a check result to a monitor: updates state, opens/resolves +// incidents on up<->down transitions, rolls up the hourly bucket, and fires +// notifications on transition. Both the server scheduler and agent-reported +// results funnel through here. +func IngestResult(monitorID string, res checker.Result) error { + ctx, cancel := monCtx() + defer cancel() + + m, err := GetMonitor(monitorID) + if err != nil || m == nil { + return err + } + + now := time.Now() + prev := m.State.Status + retries := m.Retries + if retries < 1 { + retries = 1 + } + + newStatus := prev + fails := m.State.Fails + if res.Up { + fails = 0 + newStatus = models.StatusUp + } else { + fails++ + if fails >= retries { + newStatus = models.StatusDown + } else if prev == "" || prev == models.StatusPending { + newStatus = models.StatusPending + } + } + + state := bson.M{ + "state.status": newStatus, + "state.last_check_at": now, + "state.latency_ms": res.LatencyMs, + "state.message": res.Message, + "state.fails": fails, + } + if res.CertExpiry != nil { + state["state.cert_expiry_at"] = *res.CertExpiry + } + if _, err := db.Col("monitors").UpdateOne(ctx, bson.M{"monitor_id": monitorID}, bson.M{"$set": state}); err != nil { + return err + } + + // Hourly rollup. + bucket := now.Truncate(time.Hour) + up := 0 + if res.Up { + up = 1 + } + db.Col("monitor_rollups").UpdateOne(ctx, + bson.M{"monitor_id": monitorID, "period_start": bucket}, + bson.M{"$inc": bson.M{"checks": 1, "up_count": up, "sum_latency": int64(res.LatencyMs)}}, + options.UpdateOne().SetUpsert(true)) + + // Transition handling. + if newStatus != prev { + switch newStatus { + case models.StatusDown: + inc := models.Incident{ + IncidentID: uuid.NewString(), + MonitorID: monitorID, + StartedAt: now, + Cause: res.Message, + } + db.Col("incidents").InsertOne(ctx, inc) + notifyTransition(m, newStatus, res.Message) + case models.StatusUp: + if prev == models.StatusDown { + db.Col("incidents").UpdateOne(ctx, + bson.M{"monitor_id": monitorID, "resolved_at": nil}, + bson.M{"$set": bson.M{"resolved_at": now}}) + notifyTransition(m, newStatus, res.Message) + } + } + } + return nil +} + +// notifyTransition dispatches notifications on an up<->down transition. Wired up +// in Task 13 (P3); a no-op until then. +func notifyTransition(m *models.Monitor, newStatus, message string) { + // TODO(P3): resolve m.ChannelIDs and dispatch via the notify package, + // honouring a resend interval tracked on state.last_notified_at. +}