feat: API monitoring (M16)

api/cmd/yuva/main.go

@@ -54,6 +54,7 @@ const usageText = `usage:
   yuva workspace broadcasts limit --workspace <id|name> --daily <n>   (0 removes the limit)
   yuva workspace broadcasts show --workspace <id|name>
   yuva workspace analytics-limit --workspace <id|name> <events|0>   (0 removes the limit)
+  yuva workspace monitor-limit --workspace <id|name> <requests|0>   (0 removes the limit)
   yuva analytics property create --workspace <id|name> --name <name> --domain <host>... [--inbox <id|slug>] [--platform web|ios|android]... [--timezone <zone>]
       (prints the property id and its public key, tab-separated)
   yuva analytics property list --workspace <id|name>
@@ -266,6 +267,9 @@ func serve(ctx context.Context, cfg config.Config, log *slog.Logger, args []stri
 	if cfg.WebhookAllowPrivate {
 		log.Warn("YUVA_WEBHOOK_ALLOW_PRIVATE is set; webhooks may reach private and loopback addresses")
 	}
+	if cfg.Analytics.Monitor.AllowPrivate {
+		log.Warn("YUVA_MONITOR_ALLOW_PRIVATE is set; uptime checks may call private and loopback addresses")
+	}
 	if cfg.MigrationAllowPrivate {
 		log.Warn("YUVA_MIGRATION_ALLOW_PRIVATE is set; migrations may connect to mail servers on private and loopback addresses")
 	}
@@ -274,6 +278,8 @@ func serve(ctx context.Context, cfg config.Config, log *slog.Logger, args []stri
 	}
 	if !cfg.Analytics.On {
 		log.Info("YUVA_ANALYTICS is off; analytics endpoints, the script and collection are not served")
+	} else if !cfg.Analytics.Monitor.On {
+		log.Info("YUVA_MONITOR is off; API properties and request records are not served")
 	}
 	if cfg.IngressAcceptV1 {
 		log.Warn("YUVA_INGRESS_ACCEPT_V1 is set; /ingress/email accepts v1 signatures, which do not cover the envelope sender")
@@ -285,7 +291,11 @@ func serve(ctx context.Context, cfg config.Config, log *slog.Logger, args []stri
 	if cfg.Analytics.On {
 		analyticsWorkers = cfg.Analytics.Workers
 	}
-	queue, err := jobs.New(st.Pool, st.Queries, log, cfg.BroadcastWorkers, cfg.MigrationWorkers, analyticsWorkers, srv.AddWorkers)
+	monitorWorkers := 0
+	if cfg.Analytics.On && cfg.Analytics.Monitor.On {
+		monitorWorkers = cfg.Analytics.Monitor.Workers
+	}
+	queue, err := jobs.New(st.Pool, st.Queries, log, cfg.BroadcastWorkers, cfg.MigrationWorkers, analyticsWorkers, monitorWorkers, srv.AddWorkers)
 	if err != nil {
 		return fmt.Errorf("job queue: %w", err)
 	}
@@ -332,6 +342,9 @@ func analyticsSettings(cfg config.Config) api.AnalyticsSettings {
 		Off: !a.On, Buffer: a.Buffer, VisitCache: a.VisitCache, SaltGrace: a.SaltGrace, MonthlyEvents: a.MonthlyEvents,
 		PropertyEventsPerMinute: a.PropertyEventsPerMinute, PropertiesPerWorkspace: a.PropertiesPerWorkspace,
 		RawDays: a.RawDays, RollupTop: a.RollupTop, GeoIP: a.GeoIP, GeoIPDir: a.GeoIPDir, GeoIPFile: a.GeoIPFile,
+		Monitor: api.MonitorSettings{Off: !a.Monitor.On, MonthlyRequests: a.Monitor.MonthlyRequests, RecordsPerMinute: a.Monitor.RecordsPerMinute,
+			BufferKeys: a.Monitor.BufferKeys, MinuteDays: a.Monitor.MinuteDays, Checks: a.Monitor.Checks, MinInterval: a.Monitor.MinInterval,
+			AllowPrivate: a.Monitor.AllowPrivate, CheckDays: a.Monitor.CheckDays},
 	}
 }
 

api/cmd/yuva/operator.go

@@ -34,7 +34,8 @@ const (
 		"       yuva workspace broadcasts approve|deny|suspend --workspace <id|name> [--reason <text>]\n" +
 		"       yuva workspace broadcasts limit --workspace <id|name> --daily <n>   (0 removes the limit)\n" +
 		"       yuva workspace broadcasts show --workspace <id|name>\n" +
-		"       yuva workspace analytics-limit --workspace <id|name> <events|0>   (0 removes the limit)"
+		"       yuva workspace analytics-limit --workspace <id|name> <events|0>   (0 removes the limit)\n" +
+		"       yuva workspace monitor-limit --workspace <id|name> <requests|0>   (0 removes the limit)"
 	analyticsUsage = "usage: yuva analytics property create --workspace <id|name> --name <name> --domain <host>... [--inbox <id|slug>] [--platform web|ios|android]... [--timezone <zone>]\n" +
 		"       yuva analytics property list --workspace <id|name>\n" +
 		"       yuva analytics geoip-update"
@@ -68,6 +69,9 @@ func operatorCLI(ctx context.Context, cfg config.Config, cmd string, args []stri
 		if len(args) > 0 && args[0] == "analytics-limit" {
 			return analyticsLimitCommand(ctx, st, srv, args[1:], os.Stderr)
 		}
+		if len(args) > 0 && args[0] == "monitor-limit" {
+			return monitorLimitCommand(ctx, st, srv, args[1:], os.Stderr)
+		}
 		return workspaceCommand(ctx, st, srv, args, os.Stderr)
 	case "person":
 		return personCommand(ctx, srv, args, os.Stderr)
@@ -516,6 +520,35 @@ func analyticsLimitCommand(ctx context.Context, st *store.Store, srv *api.Server
 	return nil
 }
 
+func monitorLimitCommand(ctx context.Context, st *store.Store, srv *api.Server, args []string, stderr io.Writer) error {
+	fs := flag.NewFlagSet("workspace monitor-limit", flag.ContinueOnError)
+	fs.SetOutput(stderr)
+	workspace := fs.String("workspace", "", "workspace id or exact name")
+	if err := fs.Parse(args); err != nil {
+		return err
+	}
+	if fs.NArg() != 1 {
+		return errors.New(workspaceUsage)
+	}
+	requests, err := strconv.ParseInt(fs.Arg(0), 10, 64)
+	if err != nil || requests < 0 {
+		return errors.New("the limit must be a number of requests, 0 for no limit")
+	}
+	ws, err := api.FindWorkspace(ctx, st, *workspace)
+	if err != nil {
+		return err
+	}
+	if err := srv.OperatorLimitMonitor(ctx, ws.ID, requests); err != nil {
+		return err
+	}
+	if requests == 0 {
+		fmt.Fprintf(stderr, "workspace %q (%s): API request records are not limited\n", ws.Name, ws.ID)
+		return nil
+	}
+	fmt.Fprintf(stderr, "workspace %q (%s): at most %d API request records a month\n", ws.Name, ws.ID, requests)
+	return nil
+}
+
 func analyticsCommand(ctx context.Context, st *store.Store, srv *api.Server, args []string, stdout, stderr io.Writer) error {
 	if len(args) < 2 || args[0] != "property" {
 		return errors.New(analyticsUsage)

api/go.mod

@@ -38,6 +38,7 @@ require (
 	golang.org/x/net v0.60.0
 	golang.org/x/text v0.42.0
 	golang.org/x/tools v0.50.0
+	google.golang.org/protobuf v1.36.12
 )
 
 require (
@@ -148,7 +149,6 @@ require (
 	google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20260831171406-18b4a7587f8a // indirect
 	google.golang.org/grpc v1.83.2 // indirect
-	google.golang.org/protobuf v1.36.12 // indirect
 	gopkg.in/ini.v1 v1.67.3 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect

api/go.sum

@@ -423,8 +423,6 @@ golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg=
 golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk=
 golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44=
 golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM=
-golang.org/x/net v0.59.0 h1:5zfYln+w5XCxwrnMMJPufRgNoXEaGxl0wo5GqPXyues=
-golang.org/x/net v0.59.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg=
 golang.org/x/net v0.60.0 h1:79p50tfZlm0J9YfoDsSi639qSXNGVwEzOPLCxM2FsYU=
 golang.org/x/net v0.60.0/go.mod h1:2DA/G1UfVbCpQPeWTmMPGY7Cs2PkBkwu743bVX5PIVg=
 golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=

api/internal/api/analytics_apps.go

@@ -285,6 +285,9 @@ func (s *Server) SendAnalyticsEvents(ctx context.Context, req oas.SendAnalyticsE
 	if err != nil {
 		return nil, err
 	}
+	if err := siteProperty(prop); err != nil {
+		return nil, err
+	}
 	if req.Body == nil || len(req.Body.Events) == 0 || len(req.Body.Events) > maxAnalyticsBatch {
 		return nil, errValidation("events must hold 1 to 100 events")
 	}

api/internal/api/analytics_collect.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/productdevbook/yuva/api/internal/analytics"
 	"github.com/productdevbook/yuva/api/internal/mail"
+	"github.com/productdevbook/yuva/api/internal/monitor"
 	"github.com/productdevbook/yuva/api/internal/oas"
 	"github.com/productdevbook/yuva/api/internal/realtime"
 	"github.com/productdevbook/yuva/api/internal/store"
@@ -72,6 +73,7 @@ type AnalyticsSettings struct {
 	GeoIPFile string
 	// GeoIPURL replaces DB-IP's download address, for tests.
 	GeoIPURL func(month time.Time) string
+	Monitor  MonitorSettings
 }
 
 func analyticsSettings(a AnalyticsSettings) AnalyticsSettings {
@@ -99,6 +101,7 @@ func analyticsSettings(a AnalyticsSettings) AnalyticsSettings {
 	if a.GeoIP == "" {
 		a.GeoIP = analytics.GeoOff
 	}
+	a.Monitor = monitorSettings(a.Monitor)
 	return a
 }
 
@@ -115,6 +118,7 @@ type analyticsState struct {
 	props       *propertyCache
 	usage       *usageCache
 	limits      *rateLimiter
+	monitor     *monitorState
 }
 
 func (s *Server) newAnalytics(d Deps) *analyticsState {
@@ -129,13 +133,19 @@ func (s *Server) newAnalytics(d Deps) *analyticsState {
 		a.geo = &analytics.GeoIP{Mode: set.GeoIP, Dir: set.GeoIPDir, File: set.GeoIPFile, URL: set.GeoIPURL, Now: s.now}
 	}
 	a.flusher = &analytics.Flusher{Store: st, Buffer: a.buffer, Log: s.log, Now: s.now, AfterFlush: s.afterAnalyticsFlush}
+	if !d.Analytics.Monitor.Off {
+		a.monitor = s.newMonitor(d)
+	}
 	if d.Hub != nil {
 		d.Hub.OnInvalidate(a.invalidated)
 	}
 	return a
 }
 
 func (a *analyticsState) forget(propertyID uuid.UUID) {
+	if a.monitor != nil {
+		a.monitor.agg.Forget(propertyID)
+	}
 	a.props.forget(propertyID)
 	a.answers.forget(propertyID)
 	a.resolver.Salts.Forget(propertyID)
@@ -182,17 +192,34 @@ func (a *analyticsState) invalidated(payload string) {
 		}
 		return
 	}
-	if id, err := uuid.Parse(target); err == nil {
-		a.forget(id)
+	target, erased, _ := strings.Cut(target, monitorEraseMark)
+	id, err := uuid.Parse(target)
+	if err != nil {
+		return
+	}
+	if n, err := strconv.ParseInt(erased, 10, 32); err == nil && a.monitor != nil {
+		a.monitor.agg.EraseDim(id, monitor.DimConsumer, int32(n))
 	}
+	a.forget(id)
 }
 
-// RunAnalytics writes collected events until ctx ends and then drains what is left; it returns at
-// once when analytics is off.
+// RunAnalytics writes collected events and request rows until ctx ends and then drains what is
+// left; it returns at once when analytics is off.
 func (s *Server) RunAnalytics(ctx context.Context) {
 	if s.analytics == nil {
 		return
 	}
+	if m := s.analytics.monitor; m != nil {
+		if err := m.agg.Prepare(ctx); err != nil {
+			s.log.ErrorContext(ctx, "monitor partitions", slog.Any("error", err))
+		}
+		done := make(chan struct{})
+		go func() {
+			defer close(done)
+			m.agg.Run(ctx)
+		}()
+		defer func() { <-done }()
+	}
 	if err := s.analytics.flusher.Prepare(ctx); err != nil {
 		s.log.ErrorContext(ctx, "analytics partitions", slog.Any("error", err))
 	}
@@ -371,7 +398,7 @@ func (s *Server) analyticsUsage(ctx context.Context, ws uuid.UUID) (oas.Analytic
 			out.LimitReachedAt = u.LimitReachedAt
 		}
 	}
-	return out, nil
+	return out, s.monitorUsage(ctx, ws, &out)
 }
 
 // afterAnalyticsFlush tells owners and the operator, once a month, that a workspace reached its cap.
@@ -474,7 +501,7 @@ func (s *Server) analyticsCORS(w http.ResponseWriter, r *http.Request, next http
 		w.Header().Set("Access-Control-Allow-Origin", raw)
 		w.Header().Set("Access-Control-Expose-Headers", "Retry-After")
 	}
-	key := strings.TrimPrefix(r.URL.Path, analyticsPrefix)
+	key, records := strings.CutSuffix(strings.TrimPrefix(r.URL.Path, analyticsPrefix), requestsSuffix)
 	if key == "" || strings.Contains(key, "/") || len(key) > 200 {
 		writeProblem(w, errNotFound)
 		return
@@ -485,14 +512,18 @@ func (s *Server) analyticsCORS(w http.ResponseWriter, r *http.Request, next http
 		writeProblem(w, errInternal)
 		return
 	}
-	if cp == nil {
+	if cp == nil || isAPIProperty(cp.prop) != records {
 		writeProblem(w, errNotFound)
 		return
 	}
 	if raw == "" && r.Method == http.MethodPost && cp.apps() {
 		next.ServeHTTP(w, r.WithContext(context.WithValue(r.Context(), analyticsPropertyKey, cp)))
 		return
 	}
+	if records {
+		writeProblem(w, errPropertyOrigin)
+		return
+	}
 	host := ""
 	if origin, ok := normalizeOrigin(raw); ok {
 		if u, err := url.Parse(origin); err == nil {