ev-dev-labs
diff --git a/‎.env.example‎
Lines changed: 26 additions & 0 deletions b/‎.env.example‎
Lines changed: 26 additions & 0 deletions
diff --git a/‎cmd/automation-worker/main.go‎
Lines changed: 3 additions & 32 deletions b/‎cmd/automation-worker/main.go‎
Lines changed: 3 additions & 32 deletions
diff --git a/‎cmd/automation-worker/main_test.go‎
Lines changed: 3 additions & 2 deletions b/‎cmd/automation-worker/main_test.go‎
Lines changed: 3 additions & 2 deletions
diff --git a/‎cmd/export-worker/main.go‎
Lines changed: 3 additions & 30 deletions b/‎cmd/export-worker/main.go‎
Lines changed: 3 additions & 30 deletions
diff --git a/‎cmd/export-worker/main_test.go‎
Lines changed: 3 additions & 2 deletions b/‎cmd/export-worker/main_test.go‎
Lines changed: 3 additions & 2 deletions
@@ -21,6 +21,19 @@ POSTGRES_USER=teslasync
 POSTGRES_PASSWORD=changeme
 POSTGRES_DB=teslasync
 POSTGRES_PORT=5432
+POSTGRES_MAX_CONNECTIONS=50
+# Home-lab-safe connection budget. Each worker has its own smaller pool so the
+# steady-state stack can use at most 22 connections (24 while migrations run),
+# leaving headroom below POSTGRES_MAX_CONNECTIONS for TimescaleDB maintenance.
+# Raise the server ceiling before raising any individual pool.
+DATABASE_MAX_CONNS=12
+DATABASE_MIN_CONNS=2
+NOTIFICATION_WORKER_DB_MAX_CONNS=2
+NOTIFICATION_WORKER_DB_MIN_CONNS=1
+EXPORT_WORKER_DB_MAX_CONNS=2
+EXPORT_WORKER_DB_MIN_CONNS=1
+AUTOMATION_WORKER_DB_MAX_CONNS=2
+AUTOMATION_WORKER_DB_MIN_CONNS=1
 
 # Application
 TESLASYNC_PORT=8080
@@ -83,8 +96,21 @@ FLEET_TELEMETRY_PORT=4443
 FLEET_TELEMETRY_ENABLED=false
 FLEET_TELEMETRY_HOST=
 FLEET_TELEMETRY_TOPIC_BASE=telemetry
+# Coalesce per-field MQTT messages while bounding database concurrency below
+# DATABASE_MAX_CONNS so API requests retain connection headroom.
+FLEET_TELEMETRY_BATCH_MS=100
+FLEET_TELEMETRY_BATCH_MAX_MESSAGES=256
+# Two workers are sufficient for a typical 1-2 vehicle home deployment and
+# reserve ten API-pool connections for interactive traffic. Larger fleets can
+# raise this conservatively, keeping it below DATABASE_MAX_CONNS.
+FLEET_TELEMETRY_PERSISTENCE_CONCURRENCY=2
+FLEET_TELEMETRY_PERSISTENCE_QUEUE_CAPACITY=64
+FLEET_TELEMETRY_PERSISTENCE_TIMEOUT=30s
+FLEET_TELEMETRY_SNAPSHOT_WRITE_INTERVAL=10s
 FLEET_TELEMETRY_STALE_TIMEOUT=15m
 FLEET_TELEMETRY_FALLBACK_POLL_INTERVAL=5m
+FLEET_TELEMETRY_CLEANUP_INTERVAL=2m
+FLEET_TELEMETRY_STALE_SESSION_TIMEOUT=5m
 # FLEET_TELEMETRY_TLS_CERT=./certs/server.crt
 # FLEET_TELEMETRY_TLS_KEY=./certs/server.key
 
 
@@ -32,6 +32,7 @@ import (
 	automationmodel "github.com/ev-dev-labs/teslasync/internal/models/automation"
 	tsmqtt "github.com/ev-dev-labs/teslasync/internal/mqtt"
 	"github.com/ev-dev-labs/teslasync/internal/notification"
+	healthprobe "github.com/ev-dev-labs/teslasync/internal/health"
 	"github.com/ev-dev-labs/teslasync/internal/resilience"
 	"github.com/ev-dev-labs/teslasync/internal/tesla"
 	"github.com/ev-dev-labs/teslasync/internal/tracing"
@@ -302,7 +303,8 @@ func main() {
 	// ── Health Endpoint ───────────────────────────────────────────────
 	port := healthPort()
 	healthMux := http.NewServeMux()
-	healthMux.HandleFunc("/healthz", healthHandler(db))
+	healthMux.Handle("/healthz", healthprobe.LivenessHandler())
+	healthMux.Handle("/readyz", healthprobe.ReadinessHandler(db))
 	healthMux.Handle("/metrics", promhttp.Handler())
 	go func() {
 		log.Info().Str("port", port).Msg("health endpoint listening")
@@ -356,37 +358,6 @@ func healthPort() string {
 	return port
 }
 
-// healthChecker is the minimal database surface the health endpoint needs.
-// Narrowing to this port keeps the handler unit-testable with a fake.
-type healthChecker interface {
-	Health(ctx context.Context) error
-}
-
-// healthHandler returns the /healthz handler. It responds 200 with
-// {"status":"ok"} when the checker is healthy and 503 with a JSON-encoded
-// {"status":"unhealthy","error":...} otherwise. The error message is
-// marshalled rather than string-interpolated so a checker error containing
-// quotes or newlines still produces valid, non-injectable JSON, and the
-// JSON Content-Type is set on both the success and failure paths.
-func healthHandler(checker healthChecker) http.HandlerFunc {
-	return func(w http.ResponseWriter, r *http.Request) {
-		w.Header().Set("Content-Type", "application/json")
-		if err := checker.Health(r.Context()); err != nil {
-			w.WriteHeader(http.StatusServiceUnavailable)
-			body, mErr := json.Marshal(struct {
-				Status string `json:"status"`
-				Error  string `json:"error"`
-			}{Status: "unhealthy", Error: err.Error()})
-			if mErr != nil {
-				body = []byte(`{"status":"unhealthy"}`)
-			}
-			_, _ = w.Write(body)
-			return
-		}
-		_, _ = w.Write([]byte(`{"status":"ok"}`))
-	}
-}
-
 func safePrefix(token string) string {
 	if len(token) <= 8 {
 		return token[:len(token)/2] + "***"
 
@@ -15,6 +15,7 @@ import (
 	"github.com/ev-dev-labs/teslasync/internal/database"
 	dbauto "github.com/ev-dev-labs/teslasync/internal/database/automation"
 	automationmodel "github.com/ev-dev-labs/teslasync/internal/models/automation"
+	healthprobe "github.com/ev-dev-labs/teslasync/internal/health"
 )
 
 // ── compile-time contracts ────────────────────────────────────────────────
@@ -24,7 +25,7 @@ import (
 var (
 	_ action.VariableRepo = (*variableRepoAdapter)(nil)
 	_ variableStore       = (*dbauto.AutomationVariableRepo)(nil)
-	_ healthChecker       = (*database.DB)(nil)
+	_ healthprobe.Checker = (*database.DB)(nil)
 )
 
 // ── healthPort ─────────────────────────────────────────────────────────────
@@ -171,7 +172,7 @@ func TestHealthHandler(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			fake := &fakeHealthChecker{err: tc.healthErr}
-			handler := healthHandler(fake)
+			handler := healthprobe.ReadinessHandler(fake)
 
 			req := httptest.NewRequest(http.MethodGet, "/healthz", nil)
 			rec := httptest.NewRecorder()
 
@@ -2,7 +2,6 @@ package main
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"net/http"
 	"os"
@@ -26,6 +25,7 @@ import (
 	dbbackup "github.com/ev-dev-labs/teslasync/internal/database/backup"
 	exportdb "github.com/ev-dev-labs/teslasync/internal/database/export"
 	"github.com/ev-dev-labs/teslasync/internal/export"
+	healthprobe "github.com/ev-dev-labs/teslasync/internal/health"
 	"github.com/ev-dev-labs/teslasync/internal/resilience"
 	"github.com/ev-dev-labs/teslasync/internal/tracing"
 
@@ -211,7 +211,8 @@ func main() {
 	// Health endpoint for Kubernetes probes.
 	healthPort := resolveHealthPort()
 	healthMux := http.NewServeMux()
-	healthMux.HandleFunc("/healthz", newHealthHandler(db))
+	healthMux.Handle("/healthz", healthprobe.LivenessHandler())
+	healthMux.Handle("/readyz", healthprobe.ReadinessHandler(db))
 	healthMux.Handle("/metrics", promhttp.Handler())
 	go func() {
 		log.Info().Str("port", healthPort).Msg("health endpoint listening")
@@ -292,34 +293,6 @@ func healthcheckExitCode(ctx context.Context, client *http.Client, url string) i
 	return 0
 }
 
-// healthChecker is the minimal surface newHealthHandler needs from the database
-// pool, letting the handler be exercised without a live connection.
-type healthChecker interface {
-	Health(ctx context.Context) error
-}
-
-// newHealthHandler returns the /healthz handler. It responds 200 with
-// {"status":"ok"} when the dependency is reachable and 503 with a JSON error
-// body otherwise. The error string is JSON-encoded (not string-interpolated) so
-// a driver message containing quotes cannot produce a malformed body, and
-// Content-Type is set on both paths.
-func newHealthHandler(hc healthChecker) http.HandlerFunc {
-	return func(w http.ResponseWriter, r *http.Request) {
-		w.Header().Set("Content-Type", "application/json")
-		if err := hc.Health(r.Context()); err != nil {
-			body, _ := json.Marshal(map[string]string{
-				"status": "unhealthy",
-				"error":  err.Error(),
-			})
-			w.WriteHeader(http.StatusServiceUnavailable)
-			_, _ = w.Write(body)
-			return
-		}
-		body, _ := json.Marshal(map[string]string{"status": "ok"})
-		_, _ = w.Write(body)
-	}
-}
-
 // newScheduledBackupRun builds the queued BackupRun the scheduler persists for a
 // due config. ConfigID points at the config's ID so the run is attributable to
 // its schedule, and the metadata records that a scheduled tick (not a manual
 
@@ -11,6 +11,7 @@ import (
 
 	backupmodel "github.com/ev-dev-labs/teslasync/internal/models/backup"
 
+	healthprobe "github.com/ev-dev-labs/teslasync/internal/health"
 	"github.com/rs/zerolog"
 )
 
@@ -177,7 +178,7 @@ func TestNewHealthHandler(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			handler := newHealthHandler(&fakeHealthChecker{err: tt.healthErr})
+			handler := healthprobe.ReadinessHandler(&fakeHealthChecker{err: tt.healthErr})
 			rec := httptest.NewRecorder()
 			req := httptest.NewRequest(http.MethodGet, "/healthz", nil)
 
@@ -213,7 +214,7 @@ func TestNewHealthHandler_PropagatesRequestContext(t *testing.T) {
 	const marker ctxKey = "marker"
 
 	fake := &fakeHealthChecker{}
-	handler := newHealthHandler(fake)
+	handler := healthprobe.ReadinessHandler(fake)
 
 	rec := httptest.NewRecorder()
 	req := httptest.NewRequest(http.MethodGet, "/healthz", nil)