Phase 9: Hub API and Apprise alerts (/capabilities, /ask, /notify)

Build on Phases 1-8. Adds the project side of the frozen Hub contract (central_ai_llm_master/docs/ask-contract.md) and the alert engine from AGENTS.md §4 FR-12/FR-13, §8, §11 Phase 9. - hub: internal/hub implements AskRequest/AskResponse/NotifyRequest and a deterministic, read-only Answerer. It answers screen time today/yesterday, what is blocked now, and device presence, scoped to one child or all kids. Screen time is the foreground-app total per device-day; the domain total is used only when a device reports no apps, so a report carrying both is never double-counted. - apprise: internal/apprise is a small Apprise API client (stored key or stateless URL mode, optional bearer token, typed notifications). - alert: internal/alert evaluates device state on an interval and raises one Apprise alert per transition: a limit hit (a device becomes blocked) or a device offline longer than the threshold. A never-seen device is never alerted, and the first observation seeds state without replaying an old limit. The alert is recorded before delivery, with its delivery outcome. - store: migration 005 adds `alerts` (delivery outcome, audited) and `alert_state` (per-device dedup). New usage reads from the telemetry detail table so the daily answer can be checked independently. - web: GET /capabilities, POST /ask (read-only) and POST /notify behind the shared CONTROL_HUB_TOKEN, separate from the session and admin API tokens. /notify accepts the contract's {from,level,message,person} and delivers it through Apprise (the future tamper-report path, FR-17). - cmd/control: start the alert engine (CONTROL_ALERTS_ENABLED, CONTROL_ALERT_OFFLINE_AFTER, CONTROL_ALERT_INTERVAL) and add `alerts` and `usage` subcommands for read-only inspection. - deploy/verify-phase9.sh + `make verify-phase9` prove the definition of done; version 0.7.0-phase9. Measured on .13 (2026-10-13): - GET /capabilities -> internet_use_control topics. - POST /ask "how long have the kids been on screens today?" -> "The kids have had 1h 22m of screen time today." (total_seconds=4925); the independent count from the telemetry detail table is also 4925 (app=4925, domain=4800). - A manual lock raised a delivered limit alert (id 4); an MQTT offline status raised a delivered offline alert (id 5); POST /notify delivered (id 6). All three report delivered=1 in `alerts`. Tests: gofmt, go vet ./... and go test ./... pass in golang:1.25-alpine. Co-Authored-By: Paperclip <noreply@paperclip.ing>
2026-10-13 16:41:07 +11:00
parent 4f233a8012
commit 3cf353d933
23 changed files with 2881 additions and 24 deletions
--- a/internal/web/hub_api.go
+++ b/internal/web/hub_api.go
@@ -0,0 +1,192 @@
+// The Hub contract surface (AGENTS.md §4 FR-12/FR-13, §8, §11 Phase 9).
+//
+// The project side of central_ai_llm_master/docs/ask-contract.md:
+//
+//	GET  /capabilities  -> the topics this service answers (Hub routing)
+//	POST /ask           -> a short answer about screen time and devices
+//	POST /notify        -> accept a notification and deliver it through Apprise
+//
+// All three require the shared Hub bearer token. A service-to-service call must
+// never be trusted unsigned. /ask and /capabilities are read-only: they call
+// only store read methods (AGENTS.md §10 rule 9). /notify is the one write on
+// this surface, and it only records and delivers a message; it never changes
+// device state.
+package web
+
+import (
+	"context"
+	"crypto/subtle"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"strings"
+	"time"
+
+	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/apprise"
+	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/hub"
+	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/store"
+)
+
+// hubMaxBody bounds a Hub request body. Questions and notifications are small.
+const hubMaxBody = 64 << 10 // 64 KiB
+
+// notifyResponse is this service's own reply to POST /notify. The contract does
+// not freeze a response shape for the receiver, so this is kept minimal.
+type notifyResponse struct {
+	OK    bool   `json:"ok"`
+	Error string `json:"error,omitempty"`
+}
+
+// hubAuthRequired writes the contract-shaped refusal when the Hub API has no
+// token. With no token the endpoints refuse to serve rather than trust an
+// unsigned request.
+func (s *Server) hubAuthRequired(w http.ResponseWriter) bool {
+	if !s.cfg.HubConfigured() {
+		writeHubJSON(w, http.StatusServiceUnavailable, hub.AskResponse{
+			Answer: nil,
+			Reason: "the Hub API is not configured",
+		})
+		return false
+	}
+	return true
+}
+
+// requireHub enforces the shared bearer token on the Hub endpoints.
+func (s *Server) requireHub(next http.HandlerFunc) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		if !s.hubAuthRequired(w) {
+			return
+		}
+		token, ok := bearerToken(r)
+		if !ok || subtle.ConstantTimeCompare([]byte(token), []byte(s.cfg.HubToken)) != 1 {
+			ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
+			_ = s.store.Audit(ctx, "anonymous", "hub.token_rejected", nil, r.URL.Path+" from "+clientIP(r))
+			cancel()
+			s.log.Warn("hub api rejected", "path", r.URL.Path, "remote", r.RemoteAddr)
+			w.Header().Set("WWW-Authenticate", `Bearer realm="internet-use-control"`)
+			writeHubJSON(w, http.StatusUnauthorized, hub.AskResponse{
+				Answer: nil,
+				Reason: "unauthorized",
+			})
+			return
+		}
+		next(w, r)
+	}
+}
+
+// handleCapabilities returns the routing topics for the Hub. It holds no
+// private data.
+func (s *Server) handleCapabilities(w http.ResponseWriter, _ *http.Request) {
+	writeHubJSON(w, http.StatusOK, hub.DefaultCapabilities())
+}
+
+// handleAsk answers one question from this project's read-only data. The reply
+// is always HTTP 200 with either an answer or a reason, so the Hub can relay a
+// "cannot answer" without treating the project as down.
+func (s *Server) handleAsk(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := context.WithTimeout(r.Context(), 3*time.Second)
+	defer cancel()
+
+	var req hub.AskRequest
+	dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, hubMaxBody))
+	if err := dec.Decode(&req); err != nil {
+		writeHubJSON(w, http.StatusOK, hub.AskResponse{
+			Answer: nil,
+			Reason: "the request body is not valid JSON",
+		})
+		return
+	}
+
+	started := time.Now()
+	resp := s.answerer.Answer(ctx, req)
+	s.log.Info("hub ask",
+		"asker", req.Asker,
+		"person", req.Person,
+		"source", req.Source,
+		"answered", resp.Answer != nil,
+		"duration_ms", time.Since(started).Milliseconds(),
+	)
+	writeHubJSON(w, http.StatusOK, resp)
+}
+
+// handleNotify accepts a notification in the frozen contract's shape and
+// delivers it to Sam through Apprise. It is how a tamper report from a phone
+// client (FR-17) will reach Sam without inventing a second shape. The message
+// is recorded before delivery so a delivery failure is visible.
+func (s *Server) handleNotify(w http.ResponseWriter, r *http.Request) {
+	ctx, cancel := context.WithTimeout(r.Context(), 20*time.Second)
+	defer cancel()
+
+	var req hub.NotifyRequest
+	dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, hubMaxBody))
+	if err := dec.Decode(&req); err != nil {
+		writeHubJSON(w, http.StatusBadRequest, notifyResponse{Error: "invalid JSON body"})
+		return
+	}
+
+	from := strings.TrimSpace(req.From)
+	if from == "" {
+		from = hub.ProjectName
+	}
+	level := strings.ToLower(strings.TrimSpace(req.Level))
+	switch level {
+	case "":
+		level = "info"
+	case "info", "important", "urgent":
+	default:
+		writeHubJSON(w, http.StatusBadRequest, notifyResponse{Error: "level must be info, important or urgent"})
+		return
+	}
+	message := strings.TrimSpace(req.Message)
+	if message == "" {
+		writeHubJSON(w, http.StatusBadRequest, notifyResponse{Error: "message is required"})
+		return
+	}
+	if len(message) > 500 {
+		message = message[:500]
+	}
+
+	detail := fmt.Sprintf("[%s] %s — from %s", level, message, from)
+	record, err := s.store.CreateAlert(ctx, nil, store.AlertNotify, detail)
+	if err != nil {
+		s.log.Error("notify record", "from", from, "err", err)
+		writeHubJSON(w, http.StatusInternalServerError, notifyResponse{Error: "could not record the notification"})
+		return
+	}
+
+	if !s.apprise.Configured() {
+		_ = s.store.MarkAlertDelivered(ctx, record.ID, "apprise is not configured")
+		writeHubJSON(w, http.StatusServiceUnavailable, notifyResponse{Error: "apprise is not configured"})
+		return
+	}
+	if err := s.apprise.SendType(ctx, "Internet Use Control — "+from, message, notifyType(level)); err != nil {
+		_ = s.store.MarkAlertDelivered(ctx, record.ID, err.Error())
+		s.log.Error("notify delivery", "from", from, "err", err)
+		writeHubJSON(w, http.StatusBadGateway, notifyResponse{Error: "delivery failed"})
+		return
+	}
+	if err := s.store.MarkAlertDelivered(ctx, record.ID, ""); err != nil {
+		s.log.Error("notify record", "id", record.ID, "err", err)
+	}
+	s.log.Info("notify delivered", "from", from, "level", level, "person", req.Person, "alert_id", record.ID)
+	writeHubJSON(w, http.StatusOK, notifyResponse{OK: true})
+}
+
+// notifyType maps the contract's level to an Apprise notification class.
+func notifyType(level string) string {
+	switch level {
+	case "important":
+		return apprise.TypeWarning
+	case "urgent":
+		return apprise.TypeFailure
+	default:
+		return apprise.TypeInfo
+	}
+}
+
+// writeHubJSON writes one JSON value with the given status.
+func writeHubJSON(w http.ResponseWriter, code int, v any) {
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	w.WriteHeader(code)
+	_ = json.NewEncoder(w).Encode(v)
+}
--- a/internal/web/hub_api_test.go
+++ b/internal/web/hub_api_test.go
@@ -0,0 +1,250 @@
+package web
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/auth"
+	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/config"
+	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/store"
+)
+
+const testHubToken = "test-hub-token"
+
+// newHubTestApp builds a server with a real store and explicit Hub/Apprise
+// configuration.
+func newHubTestApp(t *testing.T, hubToken, appriseBase string) (*Server, *store.Store) {
+	t.Helper()
+	st, err := store.Open(filepath.Join(t.TempDir(), "control.db"))
+	if err != nil {
+		t.Fatalf("store.Open: %v", err)
+	}
+	t.Cleanup(func() { _ = st.Close() })
+	if err := st.Migrate(context.Background()); err != nil {
+		t.Fatalf("Migrate: %v", err)
+	}
+	am, err := auth.New([]byte("0123456789abcdef0123456789abcdef"), time.Hour, false, "sam", "", "hunter2")
+	if err != nil {
+		t.Fatalf("auth.New: %v", err)
+	}
+	cfg := &config.Config{
+		AdminUsername:    "sam",
+		Version:          "test",
+		AppEnv:           "test",
+		DBPath:           ":memory:",
+		ScheduleLocation: time.UTC,
+		HubToken:         hubToken,
+		AppriseURL:       appriseBase,
+		AppriseURLs:      []string{"mailto:sam@example.com"},
+	}
+	srv, err := New(cfg, st, am, slog.New(slog.NewTextHandler(io.Discard, nil)), nil)
+	if err != nil {
+		t.Fatalf("web.New: %v", err)
+	}
+	return srv, st
+}
+
+func hubRequest(t *testing.T, h http.Handler, method, path, body, token string) *httptest.ResponseRecorder {
+	t.Helper()
+	var reader io.Reader
+	if body != "" {
+		reader = strings.NewReader(body)
+	}
+	req := httptest.NewRequest(method, path, reader)
+	if body != "" {
+		req.Header.Set("Content-Type", "application/json")
+	}
+	if token != "" {
+		req.Header.Set("Authorization", "Bearer "+token)
+	}
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+	return rec
+}
+
+func TestCapabilitiesRequiresToken(t *testing.T) {
+	srv, _ := newHubTestApp(t, testHubToken, "")
+
+	rec := hubRequest(t, srv.Handler(), http.MethodGet, "/capabilities", "", "")
+	if rec.Code != http.StatusUnauthorized {
+		t.Fatalf("unauthenticated status = %d, want 401", rec.Code)
+	}
+
+	rec = hubRequest(t, srv.Handler(), http.MethodGet, "/capabilities", "", testHubToken)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200 (%s)", rec.Code, rec.Body.String())
+	}
+	var body struct {
+		Project string   `json:"project"`
+		Topics  []string `json:"topics"`
+		People  []string `json:"people"`
+	}
+	if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	if body.Project != "internet_use_control" {
+		t.Errorf("project = %q", body.Project)
+	}
+	if len(body.Topics) == 0 {
+		t.Errorf("topics empty")
+	}
+}
+
+func TestAskRequiresTokenAndAnswers(t *testing.T) {
+	srv, st := newHubTestApp(t, testHubToken, "")
+	ctx := context.Background()
+	dev, err := st.CreateDevice(ctx, "Harry's phone", "Harry", "android", "test")
+	if err != nil {
+		t.Fatalf("CreateDevice: %v", err)
+	}
+	today := time.Now().In(time.UTC).Format("2006-01-02")
+	if _, err := st.InsertTelemetry(ctx, dev.ID, today, []store.TelemetryEntry{
+		{App: "com.spotify.music", Seconds: 3600},
+	}); err != nil {
+		t.Fatalf("InsertTelemetry: %v", err)
+	}
+
+	// No token: refused.
+	rec := hubRequest(t, srv.Handler(), http.MethodPost, "/ask",
+		`{"question":"how long have the kids been on screens today?","asker":"sam"}`, "")
+	if rec.Code != http.StatusUnauthorized {
+		t.Fatalf("unauthenticated status = %d, want 401", rec.Code)
+	}
+
+	body := `{"question":"how long have the kids been on screens today?","asker":"sam"}`
+	rec = hubRequest(t, srv.Handler(), http.MethodPost, "/ask", body, testHubToken)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200 (%s)", rec.Code, rec.Body.String())
+	}
+	var resp struct {
+		Answer *string `json:"answer"`
+		Data   struct {
+			TotalSeconds int `json:"total_seconds"`
+		} `json:"data"`
+	}
+	if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	if resp.Answer == nil || !strings.Contains(*resp.Answer, "1h 00m") {
+		t.Fatalf("answer = %v, want 1h 00m", resp.Answer)
+	}
+	if resp.Data.TotalSeconds != 3600 {
+		t.Errorf("total_seconds = %d, want 3600", resp.Data.TotalSeconds)
+	}
+}
+
+func TestAskWritesNothing(t *testing.T) {
+	srv, st := newHubTestApp(t, testHubToken, "")
+	ctx := context.Background()
+	dev, err := st.CreateDevice(ctx, "Harry's phone", "Harry", "android", "test")
+	if err != nil {
+		t.Fatalf("CreateDevice: %v", err)
+	}
+	today := time.Now().In(time.UTC).Format("2006-01-02")
+	if _, err := st.InsertTelemetry(ctx, dev.ID, today, []store.TelemetryEntry{{App: "youtube", Seconds: 600}}); err != nil {
+		t.Fatalf("InsertTelemetry: %v", err)
+	}
+
+	countRows := func() int {
+		var n int
+		if err := st.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM telemetry`).Scan(&n); err != nil {
+			t.Fatalf("count telemetry: %v", err)
+		}
+		return n
+	}
+	var auditBefore int
+	if err := st.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM audit`).Scan(&auditBefore); err != nil {
+		t.Fatalf("count audit: %v", err)
+	}
+
+	questions := []string{
+		`{"question":"how long have the kids been on screens today?","asker":"sam"}`,
+		`{"question":"what is blocked?","asker":"sam"}`,
+		`{"question":"are the devices online?","asker":"sam"}`,
+		`{"question":"what is the weather?","asker":"sam"}`,
+	}
+	for _, q := range questions {
+		rec := hubRequest(t, srv.Handler(), http.MethodPost, "/ask", q, testHubToken)
+		if rec.Code != http.StatusOK {
+			t.Fatalf("ask status = %d (%s)", rec.Code, rec.Body.String())
+		}
+	}
+	if got := countRows(); got != 1 {
+		t.Errorf("telemetry rows = %d, want 1 (ask must not write)", got)
+	}
+	var auditAfter int
+	if err := st.DB().QueryRowContext(ctx, `SELECT COUNT(*) FROM audit`).Scan(&auditAfter); err != nil {
+		t.Fatalf("count audit: %v", err)
+	}
+	if auditAfter != auditBefore {
+		t.Errorf("audit rows changed from %d to %d; ask must not write", auditBefore, auditAfter)
+	}
+}
+
+func TestNotifyDeliversThroughApprise(t *testing.T) {
+	var gotPath string
+	var gotBody map[string]any
+	stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		gotPath = r.URL.Path
+		raw, _ := io.ReadAll(r.Body)
+		_ = json.Unmarshal(raw, &gotBody)
+		w.WriteHeader(http.StatusOK)
+	}))
+	defer stub.Close()
+
+	srv, st := newHubTestApp(t, testHubToken, stub.URL)
+	rec := hubRequest(t, srv.Handler(), http.MethodPost, "/notify",
+		`{"from":"internet_use_control","level":"important","message":"3 emails need a decision.","person":"sam"}`,
+		testHubToken)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200 (%s)", rec.Code, rec.Body.String())
+	}
+	if gotPath != "/notify" {
+		t.Errorf("apprise path = %q, want /notify", gotPath)
+	}
+	if gotBody["body"] != "3 emails need a decision." {
+		t.Errorf("apprise body = %#v", gotBody)
+	}
+	if gotBody["type"] != "warning" {
+		t.Errorf("apprise type = %v, want warning", gotBody["type"])
+	}
+
+	alerts, err := st.ListAlerts(context.Background(), 10)
+	if err != nil {
+		t.Fatalf("ListAlerts: %v", err)
+	}
+	if len(alerts) != 1 || alerts[0].Kind != store.AlertNotify || !alerts[0].Delivered {
+		t.Fatalf("alerts = %#v, want one delivered notify", alerts)
+	}
+}
+
+func TestNotifyValidatesInput(t *testing.T) {
+	srv, _ := newHubTestApp(t, testHubToken, "")
+	cases := []string{
+		`{"from":"x","level":"loud","message":"hi"}`,
+		`{"from":"x","level":"info","message":""}`,
+		`not json`,
+	}
+	for _, body := range cases {
+		rec := hubRequest(t, srv.Handler(), http.MethodPost, "/notify", body, testHubToken)
+		if rec.Code != http.StatusBadRequest {
+			t.Errorf("body %q: status = %d, want 400", body, rec.Code)
+		}
+	}
+}
+
+func TestHubAPINotConfigured(t *testing.T) {
+	srv, _ := newHubTestApp(t, "", "")
+	rec := hubRequest(t, srv.Handler(), http.MethodGet, "/capabilities", "", "")
+	if rec.Code != http.StatusServiceUnavailable {
+		t.Fatalf("status = %d, want 503 when no Hub token is set", rec.Code)
+	}
+}
--- a/internal/web/server.go
+++ b/internal/web/server.go
@@ -15,8 +15,10 @@ import (
 	"sync"
 	"time"

+	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/apprise"
 	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/auth"
 	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/config"
+	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/hub"
 	"gitea.lab.audasmedia.com.au/sam/internet_use_control/internal/store"
 )

@@ -40,6 +42,11 @@ type Server struct {
 	// code cannot be tried without limit.
 	keycodeMu sync.Mutex
 	keycode   map[string]keycodeAttempts
+
+	// The Hub contract surface (Phase 9): the read-only answerer and the
+	// Apprise client that delivers alerts and accepted notifications.
+	answerer *hub.Answerer
+	apprise  *apprise.Client
 }

 // Publisher pushes device config to the broker and reports the link's health.
@@ -68,16 +75,22 @@ func New(cfg *config.Config, st *store.Store, am *auth.Manager, log *slog.Logger
 		return nil, fmt.Errorf("parse templates: %w", err)
 	}
 	return &Server{
-		cfg:     cfg,
-		store:   st,
-		auth:    am,
-		tmpl:    tmpl,
-		log:     log,
-		pub:     pub,
-		keycode: make(map[string]keycodeAttempts),
+		cfg:      cfg,
+		store:    st,
+		auth:     am,
+		tmpl:     tmpl,
+		log:      log,
+		pub:      pub,
+		keycode:  make(map[string]keycodeAttempts),
+		answerer: hub.NewAnswerer(st, cfg.ScheduleLocation),
+		apprise:  apprise.New(cfg.AppriseURL, cfg.AppriseKey, cfg.AppriseURLs, cfg.AppriseToken, 0),
 	}, nil
 }

+// Apprise exposes the delivery client so the alert engine (cmd/control) can
+// share the one configured destination with the /notify handler.
+func (s *Server) Apprise() *apprise.Client { return s.apprise }
+
 // publishConfig pushes one device's desired state to the broker, retained. A
 // publish failure never fails the HTTP request: SQLite already holds the new
 // state, and the link republishes everything when it reconnects, so the broker
@@ -131,6 +144,13 @@ func (s *Server) Handler() http.Handler {
 	// not hold that secret. It is throttled and audited instead (admin_api.go).
 	mux.HandleFunc("POST "+api+"/device/override", s.handleDeviceOverride)

+	// The Hub contract (Phase 9, FR-12/FR-13). Bearer-token auth, separate
+	// from the dashboard session and the admin API token. /ask is read-only;
+	// /notify records and delivers a message through Apprise.
+	mux.HandleFunc("GET /capabilities", s.requireHub(s.handleCapabilities))
+	mux.HandleFunc("POST /ask", s.requireHub(s.handleAsk))
+	mux.HandleFunc("POST /notify", s.requireHub(s.handleNotify))
+
 	staticSub, err := fs.Sub(staticFS, "static")
 	if err != nil {
 		panic("web: embedded static dir missing: " + err.Error())