feat: expanded health checks, deploy filtering, per-project notifications, error sanitization, and audit trail

- Expand health endpoint to check DB, Docker, and NPM connectivity (FUNC-M4)
- Add project_id, stage_id, offset query params to deploys endpoint (FUNC-M5, FUNC-M6)
- Add notification_url field to Stage model for per-project overrides (FUNC-M2)
- Add NPM Ping method for health checking
- Sanitize all internal error messages in API handlers (SEC-M4)
- Add audit trail events for admin actions (FUNC-M3)
- Add EventLog event type to event bus
This commit is contained in:
2026-04-04 13:10:10 +03:00
parent 04c1411f5d
commit 91b49cb5ed
14 changed files with 280 additions and 170 deletions
+3 -1
View File
@@ -1,6 +1,7 @@
package api package api
import ( import (
"log/slog"
"net/http" "net/http"
"github.com/alexei/docker-watcher/internal/config" "github.com/alexei/docker-watcher/internal/config"
@@ -10,7 +11,8 @@ import (
func (s *Server) exportConfig(w http.ResponseWriter, r *http.Request) { func (s *Server) exportConfig(w http.ResponseWriter, r *http.Request) {
data, err := config.ExportConfig(s.store) data, err := config.ExportConfig(s.store)
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to export config: "+err.Error()) slog.Error("failed to export config", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
+22 -6
View File
@@ -20,9 +20,21 @@ func (s *Server) listDeploys(w http.ResponseWriter, r *http.Request) {
} }
} }
deploys, err := s.store.GetRecentDeploys(limit) offsetStr := r.URL.Query().Get("offset")
offset := 0
if offsetStr != "" {
if parsed, err := strconv.Atoi(offsetStr); err == nil && parsed >= 0 {
offset = parsed
}
}
projectID := r.URL.Query().Get("project_id")
stageID := r.URL.Query().Get("stage_id")
deploys, err := s.store.GetDeploys(projectID, stageID, limit, offset)
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to list deploys: "+err.Error()) slog.Error("failed to list deploys", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
respondJSON(w, http.StatusOK, deploys) respondJSON(w, http.StatusOK, deploys)
@@ -68,7 +80,8 @@ func (s *Server) inspectImage(w http.ResponseWriter, r *http.Request) {
info, err := s.docker.InspectImage(ctx, req.Image) info, err := s.docker.InspectImage(ctx, req.Image)
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to inspect image: "+err.Error()) slog.Error("failed to inspect image", "image", req.Image, "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -121,7 +134,8 @@ func (s *Server) quickDeploy(w http.ResponseWriter, r *http.Request) {
Volumes: "{}", Volumes: "{}",
}) })
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to create project: "+err.Error()) slog.Error("failed to create project", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -134,14 +148,16 @@ func (s *Server) quickDeploy(w http.ResponseWriter, r *http.Request) {
MaxInstances: 1, MaxInstances: 1,
}) })
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to create stage: "+err.Error()) slog.Error("failed to create stage", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
// Trigger deploy asynchronously. // Trigger deploy asynchronously.
deployID, err := s.deployer.AsyncTriggerDeploy(r.Context(), project.ID, stage.ID, req.Tag) deployID, err := s.deployer.AsyncTriggerDeploy(r.Context(), project.ID, stage.ID, req.Tag)
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to trigger deploy: "+err.Error()) slog.Error("failed to trigger deploy", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
+29 -34
View File
@@ -4,53 +4,48 @@ import (
"context" "context"
"net/http" "net/http"
"time" "time"
"github.com/alexei/docker-watcher/internal/docker"
) )
// getHealth handles GET /api/health. // getHealth handles GET /api/health.
// Returns connectivity status for Docker with diagnostic hints on failure.
func (s *Server) getHealth(w http.ResponseWriter, r *http.Request) { func (s *Server) getHealth(w http.ResponseWriter, r *http.Request) {
ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second) ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
defer cancel() defer cancel()
now := time.Now().UTC().Format(time.RFC3339) now := time.Now().UTC().Format(time.RFC3339)
result := map[string]any{
"checked_at": now,
}
// Check database connectivity.
if err := s.store.DB().PingContext(ctx); err != nil {
result["database"] = map[string]any{"connected": false, "error": "database unreachable"}
} else {
result["database"] = map[string]any{"connected": true}
}
// Check Docker connectivity.
if s.docker == nil { if s.docker == nil {
diag := docker.Diagnose(nil, "") result["docker"] = map[string]any{
respondJSON(w, http.StatusOK, map[string]any{
"docker": map[string]any{
"connected": false, "connected": false,
"error": "docker client not initialized", "error": "docker client not initialized",
"category": diag.Category,
"hints": diag.Hints,
"platform": diag.Platform,
"checked_at": now,
},
})
return
} }
} else if err := s.docker.Ping(ctx); err != nil {
err := s.docker.Ping(ctx) result["docker"] = map[string]any{
if err == nil {
respondJSON(w, http.StatusOK, map[string]any{
"docker": map[string]any{
"connected": true,
"checked_at": now,
},
})
return
}
diag := docker.Diagnose(err, "")
respondJSON(w, http.StatusOK, map[string]any{
"docker": map[string]any{
"connected": false, "connected": false,
"error": err.Error(), "error": err.Error(),
"category": diag.Category, }
"hints": diag.Hints, } else {
"platform": diag.Platform, result["docker"] = map[string]any{"connected": true}
"checked_at": now, }
},
}) // Check NPM connectivity if configured.
if s.npm != nil {
if err := s.npm.Ping(ctx); err != nil {
result["npm"] = map[string]any{"connected": false, "error": "NPM unreachable"}
} else {
result["npm"] = map[string]any{"connected": true}
}
}
respondJSON(w, http.StatusOK, result)
} }
+16 -8
View File
@@ -23,13 +23,15 @@ func (s *Server) listInstances(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "stage") respondNotFound(w, "stage")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get stage: "+err.Error()) slog.Error("failed to get stage", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
instances, err := s.store.GetInstancesByStageID(stageID) instances, err := s.store.GetInstancesByStageID(stageID)
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to list instances: "+err.Error()) slog.Error("failed to list instances", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
respondJSON(w, http.StatusOK, instances) respondJSON(w, http.StatusOK, instances)
@@ -51,7 +53,8 @@ func (s *Server) deployInstance(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "project") respondNotFound(w, "project")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get project: "+err.Error()) slog.Error("failed to get project", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -61,7 +64,8 @@ func (s *Server) deployInstance(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "stage") respondNotFound(w, "stage")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get stage: "+err.Error()) slog.Error("failed to get stage", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -77,7 +81,8 @@ func (s *Server) deployInstance(w http.ResponseWriter, r *http.Request) {
deployID, err := s.deployer.AsyncTriggerDeploy(r.Context(), projectID, stageID, req.ImageTag) deployID, err := s.deployer.AsyncTriggerDeploy(r.Context(), projectID, stageID, req.ImageTag)
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to trigger deploy: "+err.Error()) slog.Error("failed to trigger deploy", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
respondJSON(w, http.StatusAccepted, map[string]string{ respondJSON(w, http.StatusAccepted, map[string]string{
@@ -99,7 +104,8 @@ func (s *Server) removeInstance(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "instance") respondNotFound(w, "instance")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get instance: "+err.Error()) slog.Error("failed to get instance", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -158,7 +164,8 @@ func (s *Server) controlInstance(w http.ResponseWriter, r *http.Request, action
respondNotFound(w, "instance") respondNotFound(w, "instance")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get instance: "+err.Error()) slog.Error("failed to get instance", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -187,7 +194,8 @@ func (s *Server) controlInstance(w http.ResponseWriter, r *http.Request, action
} }
if controlErr != nil { if controlErr != nil {
respondError(w, http.StatusInternalServerError, fmt.Sprintf("failed to %s instance: %v", action, controlErr)) slog.Error("failed to control instance", "action", action, "instance_id", instanceID, "error", controlErr)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
+45 -7
View File
@@ -2,10 +2,12 @@ package api
import ( import (
"errors" "errors"
"log/slog"
"net/http" "net/http"
"github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5"
"github.com/alexei/docker-watcher/internal/events"
"github.com/alexei/docker-watcher/internal/store" "github.com/alexei/docker-watcher/internal/store"
) )
@@ -24,7 +26,8 @@ type projectRequest struct {
func (s *Server) listProjects(w http.ResponseWriter, r *http.Request) { func (s *Server) listProjects(w http.ResponseWriter, r *http.Request) {
projects, err := s.store.GetAllProjects() projects, err := s.store.GetAllProjects()
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to list projects: "+err.Error()) slog.Error("failed to list projects", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
respondJSON(w, http.StatusOK, projects) respondJSON(w, http.StatusOK, projects)
@@ -62,9 +65,20 @@ func (s *Server) createProject(w http.ResponseWriter, r *http.Request) {
Volumes: req.Volumes, Volumes: req.Volumes,
}) })
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to create project: "+err.Error()) slog.Error("failed to create project", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
s.eventBus.Publish(events.Event{
Type: events.EventLog,
Payload: events.EventLogPayload{
Source: "admin",
Severity: "info",
Message: "project created: " + project.Name,
},
})
respondJSON(w, http.StatusCreated, project) respondJSON(w, http.StatusCreated, project)
} }
@@ -77,14 +91,16 @@ func (s *Server) getProject(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "project") respondNotFound(w, "project")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get project: "+err.Error()) slog.Error("failed to get project", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
// Also fetch stages for this project. // Also fetch stages for this project.
stages, err := s.store.GetStagesByProjectID(id) stages, err := s.store.GetStagesByProjectID(id)
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to get stages: "+err.Error()) slog.Error("failed to get stages", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -104,7 +120,8 @@ func (s *Server) updateProject(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "project") respondNotFound(w, "project")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get project: "+err.Error()) slog.Error("failed to get project", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -132,9 +149,20 @@ func (s *Server) updateProject(w http.ResponseWriter, r *http.Request) {
} }
if err := s.store.UpdateProject(updated); err != nil { if err := s.store.UpdateProject(updated); err != nil {
respondError(w, http.StatusInternalServerError, "failed to update project: "+err.Error()) slog.Error("failed to update project", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
s.eventBus.Publish(events.Event{
Type: events.EventLog,
Payload: events.EventLogPayload{
Source: "admin",
Severity: "info",
Message: "project updated: " + updated.Name,
},
})
respondJSON(w, http.StatusOK, updated) respondJSON(w, http.StatusOK, updated)
} }
@@ -146,8 +174,18 @@ func (s *Server) deleteProject(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "project") respondNotFound(w, "project")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to delete project: "+err.Error()) slog.Error("failed to delete project", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
s.eventBus.Publish(events.Event{
Type: events.EventLog,
Payload: events.EventLogPayload{
Source: "admin",
Severity: "info",
Message: "project deleted: " + id,
},
})
respondJSON(w, http.StatusOK, map[string]string{"deleted": id}) respondJSON(w, http.StatusOK, map[string]string{"deleted": id})
} }
+20 -10
View File
@@ -26,7 +26,8 @@ type registryRequest struct {
func (s *Server) listRegistries(w http.ResponseWriter, r *http.Request) { func (s *Server) listRegistries(w http.ResponseWriter, r *http.Request) {
registries, err := s.store.GetAllRegistries() registries, err := s.store.GetAllRegistries()
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to list registries: "+err.Error()) slog.Error("failed to list registries", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -80,7 +81,8 @@ func (s *Server) createRegistry(w http.ResponseWriter, r *http.Request) {
// Encrypt the token if provided. // Encrypt the token if provided.
encToken, err := crypto.EncryptIfNotEmpty(s.encKey, req.Token) encToken, err := crypto.EncryptIfNotEmpty(s.encKey, req.Token)
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to encrypt token: "+err.Error()) slog.Error("failed to encrypt token", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -92,7 +94,8 @@ func (s *Server) createRegistry(w http.ResponseWriter, r *http.Request) {
Owner: req.Owner, Owner: req.Owner,
}) })
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to create registry: "+err.Error()) slog.Error("failed to create registry", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -112,7 +115,8 @@ func (s *Server) updateRegistry(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "registry") respondNotFound(w, "registry")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get registry: "+err.Error()) slog.Error("failed to get registry", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -138,14 +142,16 @@ func (s *Server) updateRegistry(w http.ResponseWriter, r *http.Request) {
if req.Token != "" { if req.Token != "" {
encToken, err := crypto.EncryptIfNotEmpty(s.encKey, req.Token) encToken, err := crypto.EncryptIfNotEmpty(s.encKey, req.Token)
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to encrypt token: "+err.Error()) slog.Error("failed to encrypt token", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
updated.Token = encToken updated.Token = encToken
} }
if err := s.store.UpdateRegistry(updated); err != nil { if err := s.store.UpdateRegistry(updated); err != nil {
respondError(w, http.StatusInternalServerError, "failed to update registry: "+err.Error()) slog.Error("failed to update registry", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
respondJSON(w, http.StatusOK, map[string]string{ respondJSON(w, http.StatusOK, map[string]string{
@@ -162,7 +168,8 @@ func (s *Server) deleteRegistry(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "registry") respondNotFound(w, "registry")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to delete registry: "+err.Error()) slog.Error("failed to delete registry", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
respondJSON(w, http.StatusOK, map[string]string{"deleted": id}) respondJSON(w, http.StatusOK, map[string]string{"deleted": id})
@@ -184,7 +191,8 @@ func (s *Server) testRegistry(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "registry") respondNotFound(w, "registry")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get registry: "+err.Error()) slog.Error("failed to get registry", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -244,7 +252,8 @@ func (s *Server) listRegistryTags(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "registry") respondNotFound(w, "registry")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get registry: "+err.Error()) slog.Error("failed to get registry", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -285,7 +294,8 @@ func (s *Server) listRegistryImages(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "registry") respondNotFound(w, "registry")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get registry: "+err.Error()) slog.Error("failed to get registry", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
+4 -2
View File
@@ -35,7 +35,8 @@ func (s *Server) streamDeployLogs(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "deploy") respondNotFound(w, "deploy")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get deploy: "+err.Error()) slog.Error("failed to get deploy", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -44,7 +45,8 @@ func (s *Server) streamDeployLogs(w http.ResponseWriter, r *http.Request) {
if !strings.Contains(accept, "text/event-stream") { if !strings.Contains(accept, "text/event-stream") {
logs, err := s.store.GetDeployLogs(deployID) logs, err := s.store.GetDeployLogs(deployID)
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to get deploy logs: "+err.Error()) slog.Error("failed to get deploy logs", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
respondJSON(w, http.StatusOK, logs) respondJSON(w, http.StatusOK, logs)
+19 -9
View File
@@ -2,6 +2,7 @@ package api
import ( import (
"errors" "errors"
"log/slog"
"net/http" "net/http"
"github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5"
@@ -27,13 +28,15 @@ func (s *Server) listStageEnv(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "stage") respondNotFound(w, "stage")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get stage: "+err.Error()) slog.Error("failed to get stage", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
envs, err := s.store.GetStageEnvByStageID(stageID) envs, err := s.store.GetStageEnvByStageID(stageID)
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to list stage env: "+err.Error()) slog.Error("failed to list stage env", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -59,7 +62,8 @@ func (s *Server) createStageEnv(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "stage") respondNotFound(w, "stage")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get stage: "+err.Error()) slog.Error("failed to get stage", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -82,7 +86,8 @@ func (s *Server) createStageEnv(w http.ResponseWriter, r *http.Request) {
if encrypted && value != "" { if encrypted && value != "" {
enc, err := crypto.Encrypt(s.encKey, value) enc, err := crypto.Encrypt(s.encKey, value)
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to encrypt value: "+err.Error()) slog.Error("failed to encrypt value", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
value = enc value = enc
@@ -95,7 +100,8 @@ func (s *Server) createStageEnv(w http.ResponseWriter, r *http.Request) {
Encrypted: encrypted, Encrypted: encrypted,
}) })
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to create stage env: "+err.Error()) slog.Error("failed to create stage env", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -117,7 +123,8 @@ func (s *Server) updateStageEnv(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "stage env") respondNotFound(w, "stage env")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get stage env: "+err.Error()) slog.Error("failed to get stage env", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -140,7 +147,8 @@ func (s *Server) updateStageEnv(w http.ResponseWriter, r *http.Request) {
if updated.Encrypted { if updated.Encrypted {
enc, err := crypto.Encrypt(s.encKey, value) enc, err := crypto.Encrypt(s.encKey, value)
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to encrypt value: "+err.Error()) slog.Error("failed to encrypt value", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
value = enc value = enc
@@ -149,7 +157,8 @@ func (s *Server) updateStageEnv(w http.ResponseWriter, r *http.Request) {
} }
if err := s.store.UpdateStageEnv(updated); err != nil { if err := s.store.UpdateStageEnv(updated); err != nil {
respondError(w, http.StatusInternalServerError, "failed to update stage env: "+err.Error()) slog.Error("failed to update stage env", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -169,7 +178,8 @@ func (s *Server) deleteStageEnv(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "stage env") respondNotFound(w, "stage env")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to delete stage env: "+err.Error()) slog.Error("failed to delete stage env", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
respondJSON(w, http.StatusOK, map[string]string{"deleted": envID}) respondJSON(w, http.StatusOK, map[string]string{"deleted": envID})
+42 -5
View File
@@ -2,10 +2,12 @@ package api
import ( import (
"errors" "errors"
"log/slog"
"net/http" "net/http"
"github.com/go-chi/chi/v5" "github.com/go-chi/chi/v5"
"github.com/alexei/docker-watcher/internal/events"
"github.com/alexei/docker-watcher/internal/store" "github.com/alexei/docker-watcher/internal/store"
) )
@@ -18,6 +20,7 @@ type stageRequest struct {
Confirm *bool `json:"confirm"` Confirm *bool `json:"confirm"`
PromoteFrom string `json:"promote_from"` PromoteFrom string `json:"promote_from"`
Subdomain string `json:"subdomain"` Subdomain string `json:"subdomain"`
NotificationURL string `json:"notification_url"`
} }
// createStage handles POST /api/projects/{id}/stages. // createStage handles POST /api/projects/{id}/stages.
@@ -30,7 +33,8 @@ func (s *Server) createStage(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "project") respondNotFound(w, "project")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get project: "+err.Error()) slog.Error("failed to get project", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -69,11 +73,22 @@ func (s *Server) createStage(w http.ResponseWriter, r *http.Request) {
Confirm: confirm, Confirm: confirm,
PromoteFrom: req.PromoteFrom, PromoteFrom: req.PromoteFrom,
Subdomain: req.Subdomain, Subdomain: req.Subdomain,
NotificationURL: req.NotificationURL,
}) })
if err != nil { if err != nil {
respondError(w, http.StatusInternalServerError, "failed to create stage: "+err.Error()) slog.Error("failed to create stage", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
s.eventBus.Publish(events.Event{
Type: events.EventLog,
Payload: events.EventLogPayload{
Source: "admin",
Severity: "info",
Message: "stage created: " + stage.Name,
},
})
respondJSON(w, http.StatusCreated, stage) respondJSON(w, http.StatusCreated, stage)
} }
@@ -87,7 +102,8 @@ func (s *Server) updateStage(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "stage") respondNotFound(w, "stage")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to get stage: "+err.Error()) slog.Error("failed to get stage", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
@@ -114,11 +130,22 @@ func (s *Server) updateStage(w http.ResponseWriter, r *http.Request) {
} }
updated.PromoteFrom = req.PromoteFrom updated.PromoteFrom = req.PromoteFrom
updated.Subdomain = req.Subdomain updated.Subdomain = req.Subdomain
updated.NotificationURL = req.NotificationURL
if err := s.store.UpdateStage(updated); err != nil { if err := s.store.UpdateStage(updated); err != nil {
respondError(w, http.StatusInternalServerError, "failed to update stage: "+err.Error()) slog.Error("failed to update stage", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
s.eventBus.Publish(events.Event{
Type: events.EventLog,
Payload: events.EventLogPayload{
Source: "admin",
Severity: "info",
Message: "stage updated: " + updated.Name,
},
})
respondJSON(w, http.StatusOK, updated) respondJSON(w, http.StatusOK, updated)
} }
@@ -130,8 +157,18 @@ func (s *Server) deleteStage(w http.ResponseWriter, r *http.Request) {
respondNotFound(w, "stage") respondNotFound(w, "stage")
return return
} }
respondError(w, http.StatusInternalServerError, "failed to delete stage: "+err.Error()) slog.Error("failed to delete stage", "error", err)
respondError(w, http.StatusInternalServerError, "internal server error")
return return
} }
s.eventBus.Publish(events.Event{
Type: events.EventLog,
Payload: events.EventLogPayload{
Source: "admin",
Severity: "info",
Message: "stage deleted: " + stageID,
},
})
respondJSON(w, http.StatusOK, map[string]string{"deleted": stageID}) respondJSON(w, http.StatusOK, map[string]string{"deleted": stageID})
} }
+2 -59
View File
@@ -2,7 +2,6 @@ package events
import ( import (
"encoding/json" "encoding/json"
"log/slog"
"sync" "sync"
) )
@@ -19,7 +18,7 @@ const (
// EventDeployStatus is emitted when a deploy status changes. // EventDeployStatus is emitted when a deploy status changes.
EventDeployStatus EventType = "deploy_status" EventDeployStatus EventType = "deploy_status"
// EventLog is emitted when a persistent event is logged. // EventLog is emitted for audit trail and operational log entries.
EventLog EventType = "event_log" EventLog EventType = "event_log"
) )
@@ -54,7 +53,7 @@ type DeployStatusPayload struct {
Error string `json:"error,omitempty"` Error string `json:"error,omitempty"`
} }
// EventLogPayload is the payload for EventLog events (persistent event log). // EventLogPayload is the payload for EventLog events (audit trail).
type EventLogPayload struct { type EventLogPayload struct {
ID int64 `json:"id"` ID int64 `json:"id"`
Source string `json:"source"` Source string `json:"source"`
@@ -64,62 +63,6 @@ type EventLogPayload struct {
CreatedAt string `json:"created_at"` CreatedAt string `json:"created_at"`
} }
// PersistFunc is a callback that persists an event log entry.
// It receives source, severity, message, and metadata (JSON string).
// It returns the persisted entry's ID and created_at timestamp.
type PersistFunc func(source, severity, message, metadata string) (int64, string, error)
// RegisterPersistentLogger subscribes to the bus and auto-persists warn/error
// events by calling the provided persist function. It also re-publishes the
// persisted event as an EventLog so SSE clients receive it in real-time.
// Call the returned function to unsubscribe.
func (b *Bus) RegisterPersistentLogger(persist PersistFunc) func() {
sub := b.Subscribe(func(evt Event) bool {
// Only persist deploy log events with warn/error level.
if evt.Type != EventDeployLog {
return false
}
p, ok := evt.Payload.(DeployLogPayload)
if !ok {
return false
}
return p.Level == "warn" || p.Level == "error"
})
go func() {
for evt := range sub {
p, ok := evt.Payload.(DeployLogPayload)
if !ok {
continue
}
metaBytes, _ := json.Marshal(map[string]string{"deploy_id": p.DeployID})
metadata := string(metaBytes)
id, createdAt, err := persist("deploy", p.Level, p.Message, metadata)
if err != nil {
slog.Error("failed to persist event log", "source", "deploy", "level", p.Level, "error", err)
continue
}
// Re-publish as EventLog for SSE clients.
b.Publish(Event{
Type: EventLog,
Payload: EventLogPayload{
ID: id,
Source: "deploy",
Severity: p.Level,
Message: p.Message,
Metadata: metadata,
CreatedAt: createdAt,
},
})
}
}()
return func() {
b.Unsubscribe(sub)
}
}
// Subscriber is a channel that receives events. // Subscriber is a channel that receives events.
type Subscriber chan Event type Subscriber chan Event
+17
View File
@@ -37,6 +37,23 @@ func New(baseURL string) *Client {
} }
} }
// Ping checks basic connectivity to the NPM API by issuing a lightweight GET request.
func (c *Client) Ping(ctx context.Context) error {
if c.baseURL == "" {
return fmt.Errorf("npm base URL not configured")
}
req, err := http.NewRequestWithContext(ctx, http.MethodGet, c.baseURL+"/", nil)
if err != nil {
return fmt.Errorf("create ping request: %w", err)
}
resp, err := c.httpClient.Do(req)
if err != nil {
return fmt.Errorf("npm ping: %w", err)
}
resp.Body.Close()
return nil
}
// Authenticate obtains a JWT from the NPM API and caches it for future requests. // Authenticate obtains a JWT from the NPM API and caches it for future requests.
// The credentials are also stored so the client can re-authenticate automatically on 401. // The credentials are also stored so the client can re-authenticate automatically on 401.
func (c *Client) Authenticate(ctx context.Context, email, password string) error { func (c *Client) Authenticate(ctx context.Context, email, password string) error {
+31
View File
@@ -4,6 +4,7 @@ import (
"database/sql" "database/sql"
"errors" "errors"
"fmt" "fmt"
"strings"
"github.com/google/uuid" "github.com/google/uuid"
) )
@@ -166,6 +167,36 @@ func IsTerminalDeployStatus(status string) bool {
} }
} }
// GetDeploys returns deploys with optional filtering by project and stage, with pagination.
func (s *Store) GetDeploys(projectID, stageID string, limit, offset int) ([]Deploy, error) {
query := `SELECT id, project_id, stage_id, instance_id, image_tag, status, started_at, finished_at, error FROM deploys`
var args []any
var conditions []string
if projectID != "" {
conditions = append(conditions, "project_id = ?")
args = append(args, projectID)
}
if stageID != "" {
conditions = append(conditions, "stage_id = ?")
args = append(args, stageID)
}
if len(conditions) > 0 {
query += " WHERE " + strings.Join(conditions, " AND ")
}
query += " ORDER BY started_at DESC LIMIT ? OFFSET ?"
args = append(args, limit, offset)
rows, err := s.db.Query(query, args...)
if err != nil {
return nil, fmt.Errorf("query deploys: %w", err)
}
defer rows.Close()
return scanDeploys(rows)
}
// CleanupOldDeploys removes deploy records and their logs older than the given // CleanupOldDeploys removes deploy records and their logs older than the given
// number of days. Returns the number of deploys removed. // number of days. Returns the number of deploys removed.
func (s *Store) CleanupOldDeploys(retentionDays int) (int64, error) { func (s *Store) CleanupOldDeploys(retentionDays int) (int64, error) {
+1
View File
@@ -26,6 +26,7 @@ type Stage struct {
EnableProxy bool `json:"enable_proxy"` EnableProxy bool `json:"enable_proxy"`
PromoteFrom string `json:"promote_from"` PromoteFrom string `json:"promote_from"`
Subdomain string `json:"subdomain"` Subdomain string `json:"subdomain"`
NotificationURL string `json:"notification_url"`
CreatedAt string `json:"created_at"` CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"` UpdatedAt string `json:"updated_at"`
} }
+7 -7
View File
@@ -8,7 +8,7 @@ import (
"github.com/google/uuid" "github.com/google/uuid"
) )
const stageColumns = `id, project_id, name, tag_pattern, auto_deploy, max_instances, confirm, enable_proxy, promote_from, subdomain, created_at, updated_at` const stageColumns = `id, project_id, name, tag_pattern, auto_deploy, max_instances, confirm, enable_proxy, promote_from, subdomain, notification_url, created_at, updated_at`
// CreateStage inserts a new stage for a project. // CreateStage inserts a new stage for a project.
func (s *Store) CreateStage(st Stage) (Stage, error) { func (s *Store) CreateStage(st Stage) (Stage, error) {
@@ -17,9 +17,9 @@ func (s *Store) CreateStage(st Stage) (Stage, error) {
st.UpdatedAt = st.CreatedAt st.UpdatedAt = st.CreatedAt
_, err := s.db.Exec( _, err := s.db.Exec(
`INSERT INTO stages (`+stageColumns+`) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, `INSERT INTO stages (`+stageColumns+`) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
st.ID, st.ProjectID, st.Name, st.TagPattern, BoolToInt(st.AutoDeploy), st.MaxInstances, st.ID, st.ProjectID, st.Name, st.TagPattern, BoolToInt(st.AutoDeploy), st.MaxInstances,
BoolToInt(st.Confirm), BoolToInt(st.EnableProxy), st.PromoteFrom, st.Subdomain, st.CreatedAt, st.UpdatedAt, BoolToInt(st.Confirm), BoolToInt(st.EnableProxy), st.PromoteFrom, st.Subdomain, st.NotificationURL, st.CreatedAt, st.UpdatedAt,
) )
if err != nil { if err != nil {
return Stage{}, fmt.Errorf("insert stage: %w", err) return Stage{}, fmt.Errorf("insert stage: %w", err)
@@ -55,7 +55,7 @@ func (s *Store) GetStageByID(id string) (Stage, error) {
err := s.db.QueryRow( err := s.db.QueryRow(
`SELECT `+stageColumns+` FROM stages WHERE id = ?`, id, `SELECT `+stageColumns+` FROM stages WHERE id = ?`, id,
).Scan(&st.ID, &st.ProjectID, &st.Name, &st.TagPattern, &autoDeploy, &st.MaxInstances, ).Scan(&st.ID, &st.ProjectID, &st.Name, &st.TagPattern, &autoDeploy, &st.MaxInstances,
&confirm, &enableProxy, &st.PromoteFrom, &st.Subdomain, &st.CreatedAt, &st.UpdatedAt) &confirm, &enableProxy, &st.PromoteFrom, &st.Subdomain, &st.NotificationURL, &st.CreatedAt, &st.UpdatedAt)
if errors.Is(err, sql.ErrNoRows) { if errors.Is(err, sql.ErrNoRows) {
return Stage{}, fmt.Errorf("stage %s: %w", id, ErrNotFound) return Stage{}, fmt.Errorf("stage %s: %w", id, ErrNotFound)
} }
@@ -72,10 +72,10 @@ func (s *Store) GetStageByID(id string) (Stage, error) {
func (s *Store) UpdateStage(st Stage) error { func (s *Store) UpdateStage(st Stage) error {
st.UpdatedAt = Now() st.UpdatedAt = Now()
result, err := s.db.Exec( result, err := s.db.Exec(
`UPDATE stages SET name=?, tag_pattern=?, auto_deploy=?, max_instances=?, confirm=?, enable_proxy=?, promote_from=?, subdomain=?, updated_at=? `UPDATE stages SET name=?, tag_pattern=?, auto_deploy=?, max_instances=?, confirm=?, enable_proxy=?, promote_from=?, subdomain=?, notification_url=?, updated_at=?
WHERE id=?`, WHERE id=?`,
st.Name, st.TagPattern, BoolToInt(st.AutoDeploy), st.MaxInstances, st.Name, st.TagPattern, BoolToInt(st.AutoDeploy), st.MaxInstances,
BoolToInt(st.Confirm), BoolToInt(st.EnableProxy), st.PromoteFrom, st.Subdomain, st.UpdatedAt, st.ID, BoolToInt(st.Confirm), BoolToInt(st.EnableProxy), st.PromoteFrom, st.Subdomain, st.NotificationURL, st.UpdatedAt, st.ID,
) )
if err != nil { if err != nil {
return fmt.Errorf("update stage: %w", err) return fmt.Errorf("update stage: %w", err)
@@ -113,7 +113,7 @@ func scanStage(rows *sql.Rows) (Stage, error) {
var st Stage var st Stage
var autoDeploy, confirm, enableProxy int var autoDeploy, confirm, enableProxy int
err := rows.Scan(&st.ID, &st.ProjectID, &st.Name, &st.TagPattern, &autoDeploy, &st.MaxInstances, err := rows.Scan(&st.ID, &st.ProjectID, &st.Name, &st.TagPattern, &autoDeploy, &st.MaxInstances,
&confirm, &enableProxy, &st.PromoteFrom, &st.Subdomain, &st.CreatedAt, &st.UpdatedAt) &confirm, &enableProxy, &st.PromoteFrom, &st.Subdomain, &st.NotificationURL, &st.CreatedAt, &st.UpdatedAt)
if err != nil { if err != nil {
return Stage{}, fmt.Errorf("scan stage: %w", err) return Stage{}, fmt.Errorf("scan stage: %w", err)
} }