feat(canary): wire event + notify into runtime + create-tier rate limits

Swaps the Phase-9 directEventRecorder/mysqlEventRecorder shims for
real event.Service.Record via an eventRecorderAdapter. Constructs
notify.Service with telegram + webhook senders. Spawns the retention
loop goroutine off the main wait group; gracefulShutdown now also
notifySvc.Wait()s for in-flight notifications to drain.

Closes deferred Phase-9 items 2 (real event recorder), 3 (nested
create-tier rate limits), 5 (FingerprintRecorder wiring), 7 (drop
required tag on TurnstileResp).

- cmd/canary/main.go:
  * buildEventStack constructs telegram + webhook senders and registers
    them on notify.Service. event.Service wired with eventRepo (Store +
    StatusWriter), tokenRepo (TokenIncrementer), redis client, notifier.
  * eventRecorderAdapter and a single mysqlEventRecorder both delegate
    to event.Service.Record(ctx, t.NotifyInfo(), evt) — one path,
    same dedup + notify semantics for HTTP triggers and mysql triggers.
  * fingerprintRecorderAdapter wraps event.Repository.AttachFingerprint
    with the configured window (defaults to 5 min) so the slow-redirect
    fingerprint POST writes through.
  * spawnRetentionLoop runs event.Service.RunRetentionLoop on the
    shared wait group; configured interval + per-token limit gate it.
  * /api/tokens POST gains nested create-tier limits per spec §11.3:
    LimitCreateMin (5/min) + LimitCreateHour (20/hr) chained ahead of
    Turnstile, with their own fingerprint-keyed Redis namespaces so
    they don't share buckets with the read-tier limiter.
  * Shutdown order: server → telemetry → redis → db, then
    notifySvc.Wait() to drain in-flight goroutines, then wg.Wait()
    for retention loop + mysql listener.

- internal/config/config.go:
  * NotifyConfig: dedup_ttl, send_timeout, max_tries, max_elapsed,
    initial_interval, retention_interval, retention_limit,
    webhook_hmac_secret, telegram_api_base, fingerprint_window.
  * RateLimitConfig: create_min_rate / burst, create_hour_rate / burst.
  * Defaults + env mappings (NOTIFY_*, RATE_LIMIT_CREATE_*, plus the
    spec'd WEBHOOK_HMAC_SECRET).

- internal/token/dto.go: drop the `validate:"required"` tag on
  cf_turnstile_response. The middleware (TurnstileVerify) is the sole
  authority — it bypasses on empty SecretKey for dev mode and would
  reject empty-token submissions in prod via the Verifier's own
  ErrEmptyToken path. The validator tag was forcing tests to pass
  a placeholder string in dev mode which conflicted with the bypass.

PROJECTS/beginner/canary-token-generator/backend/cmd/canary/main.go

@@ -5,10 +5,12 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"flag"
 	"fmt"
 	"log/slog"
+	"net/http"
 	"os"
 	"os/signal"
 	"sync"
@@ -22,6 +24,9 @@ import (
 	"github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/event"
 	"github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/health"
 	"github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/middleware"
+	"github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/notify"
+	"github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/notify/telegram"
+	"github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/notify/webhook"
 	"github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/server"
 	"github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/token"
 	"github.com/CarterPerez-dev/cybersecurity-projects/canary-token-generator/backend/internal/token/generators/mysql"
@@ -86,6 +91,14 @@ func run(configPath string) error {
 	}
 	logger.Info("redis connected")
 
+	notifySvc, eventSvc := buildEventStack(
+		cfg,
+		logger,
+		eventRepo,
+		tokenRepo,
+		rdb,
+	)
+
 	genRegistry := registry.Build(registry.Config{
 		BaseURL:         cfg.Canary.BaseURL,
 		MySQLPublicHost: cfg.MySQL.PublicHost,
@@ -108,19 +121,19 @@ func run(configPath string) error {
 	healthH := health.NewHandler(db, rdb)
 	tokenH := token.NewHandler(
 		tokenSvc,
-		&directEventRecorder{
+		&eventRecorderAdapter{svc: eventSvc},
+		&fingerprintRecorderAdapter{
 			repo:   eventRepo,
-			tokens: tokenRepo,
-			logger: logger,
+			window: cfg.Notify.FingerprintWindow,
 		},
-		nil,
 		logger,
 	)
 
 	srv := mountRouter(cfg, logger, rdb, healthH, tokenH, verifier)
 
 	var wg sync.WaitGroup
-	spawnMySQLListener(ctx, cfg, logger, &wg, tokenSvc, eventRepo, tokenRepo)
+	spawnMySQLListener(ctx, cfg, logger, &wg, tokenSvc, eventSvc)
+	spawnRetentionLoop(ctx, cfg, &wg, eventSvc)
 
 	errChan := make(chan error, 1)
 	go func() { errChan <- srv.Start() }()
@@ -133,18 +146,60 @@ func run(configPath string) error {
 	}
 
 	shutdownErr := gracefulShutdown(cfg, logger, srv, telemetry, rdb, db)
+	logger.Info("waiting for in-flight notifications")
+	notifySvc.Wait()
 	wg.Wait()
 	return shutdownErr
 }
 
+func buildEventStack(
+	cfg *config.Config,
+	logger *slog.Logger,
+	eventRepo *event.Repository,
+	tokenRepo *token.Repository,
+	rdb *core.Redis,
+) (*notify.Service, *event.Service) {
+	tgSender := telegram.NewSender(telegram.Config{
+		APIBase:         cfg.Notify.TelegramAPIBase,
+		ManageURL:       cfg.Canary.ManageURL,
+		MaxTries:        cfg.Notify.MaxTries,
+		MaxElapsed:      cfg.Notify.MaxElapsed,
+		InitialInterval: cfg.Notify.InitialInterval,
+	})
+	whSender := webhook.NewSender(webhook.Config{
+		ManageURL:       cfg.Canary.ManageURL,
+		HMACSecret:      cfg.Notify.WebhookHMACSecret,
+		MaxTries:        cfg.Notify.MaxTries,
+		MaxElapsed:      cfg.Notify.MaxElapsed,
+		InitialInterval: cfg.Notify.InitialInterval,
+	})
+
+	notifySvc := notify.NewService(eventRepo,
+		notify.WithLogger(logger),
+		notify.WithSendTimeout(cfg.Notify.SendTimeout),
+	)
+	notifySvc.Register(tgSender, whSender)
+
+	eventSvc := event.NewService(
+		eventRepo,
+		tokenRepo,
+		rdb.Client,
+		notifySvc,
+		event.ServiceConfig{
+			DedupTTL: cfg.Notify.DedupTTL,
+			Logger:   logger,
+		},
+	)
+	return notifySvc, eventSvc
+}
+
 func spawnMySQLListener(
 	ctx context.Context,
 	cfg *config.Config,
 	logger *slog.Logger,
 	wg *sync.WaitGroup,
 	tokenSvc *token.Service,
-	eventRepo *event.Repository,
-	tokenRepo *token.Repository,
+	eventSvc *event.Service,
 ) {
 	if !cfg.MySQL.Enabled {
 		return
@@ -154,11 +209,7 @@ func spawnMySQLListener(
 		defer wg.Done()
 		handler := mysql.NewHandler(
 			&mysqlTokenLookup{svc: tokenSvc},
-			&mysqlEventRecorder{
-				repo:   eventRepo,
-				tokens: tokenRepo,
-				logger: logger,
-			},
+			&eventRecorderAdapter{svc: eventSvc},
 		)
 		if mErr := mysql.Run(ctx, cfg.MySQL.Addr, handler); mErr != nil {
 			logger.Error("mysql server error", "error", mErr)
@@ -166,6 +217,26 @@ func spawnMySQLListener(
 	}()
 }
 
+func spawnRetentionLoop(
+	ctx context.Context,
+	cfg *config.Config,
+	wg *sync.WaitGroup,
+	eventSvc *event.Service,
+) {
+	if cfg.Notify.RetentionInterval <= 0 || cfg.Notify.RetentionLimit <= 0 {
+		return
+	}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		eventSvc.RunRetentionLoop(
+			ctx,
+			cfg.Notify.RetentionInterval,
+			cfg.Notify.RetentionLimit,
+		)
+	}()
+}
+
 func mountRouter(
 	cfg *config.Config,
 	logger *slog.Logger,
@@ -189,6 +260,29 @@ func mountRouter(
 	healthH.RegisterRoutes(r)
 	tokenH.RegisterTriggerRoutes(r)
 
+	createMin := middleware.NewRateLimiter(
+		rdb.Client,
+		middleware.RateLimitConfig{
+			Limit: middleware.PerMinute(
+				cfg.RateLimit.CreateMinRate,
+				cfg.RateLimit.CreateMinBurst,
+			),
+			KeyFunc:  keyByCreateMin,
+			FailOpen: true,
+		},
+	).Handler
+	createHour := middleware.NewRateLimiter(
+		rdb.Client,
+		middleware.RateLimitConfig{
+			Limit: middleware.PerHour(
+				cfg.RateLimit.CreateHourRate,
+				cfg.RateLimit.CreateHourBurst,
+			),
+			KeyFunc:  keyByCreateHour,
+			FailOpen: true,
+		},
+	).Handler
+
 	r.Route("/api", func(api chi.Router) {
 		api.Use(middleware.CORS(cfg.CORS))
 		api.Use(
@@ -202,13 +296,24 @@ func mountRouter(
 			}).Handler,
 		)
 		api.Get("/tokens/types", tokenH.GetTypes)
-		api.With(middleware.TurnstileVerify(verifier)).
-			Post("/tokens", tokenH.CreateToken)
+		api.With(
+			createMin,
+			createHour,
+			middleware.TurnstileVerify(verifier),
+		).Post("/tokens", tokenH.CreateToken)
 	})
 
 	return srv
 }
 
+func keyByCreateMin(r *http.Request) string {
+	return "ratelimit:create:min:" + middleware.ExtractFingerprint(r)
+}
+
+func keyByCreateHour(r *http.Request) string {
+	return "ratelimit:create:hour:" + middleware.ExtractFingerprint(r)
+}
+
 func gracefulShutdown(
 	cfg *config.Config,
 	logger *slog.Logger,
@@ -291,25 +396,35 @@ func (a registryAdapter) Get(t token.Type) (token.Generator, bool) {
 	return g, ok
 }
 
-type directEventRecorder struct {
-	repo   *event.Repository
-	tokens *token.Repository
-	logger *slog.Logger
+type eventRecorderAdapter struct {
+	svc *event.Service
 }
 
-func (d *directEventRecorder) Record(
+func (a *eventRecorderAdapter) Record(
 	ctx context.Context,
 	t *token.Token,
 	evt *event.Event,
 ) error {
-	if err := d.repo.Insert(ctx, evt); err != nil {
-		return fmt.Errorf("insert event: %w", err)
-	}
-	if err := d.tokens.IncrementTriggerCount(ctx, t.ID); err != nil {
-		d.logger.WarnContext(ctx, "increment trigger count",
-			"error", err, "token_id", t.ID)
-	}
-	return nil
+	return a.svc.Record(ctx, t.NotifyInfo(), evt)
+}
+
+type fingerprintRecorderAdapter struct {
+	repo   *event.Repository
+	window time.Duration
+}
+
+func (f *fingerprintRecorderAdapter) AttachFingerprint(
+	ctx context.Context,
+	tokenID, sourceIP string,
+	fingerprint json.RawMessage,
+) error {
+	return f.repo.AttachFingerprint(
+		ctx,
+		tokenID,
+		sourceIP,
+		fingerprint,
+		f.window,
+	)
 }
 
 type mysqlTokenLookup struct{ svc *token.Service }
@@ -320,24 +435,3 @@ func (m *mysqlTokenLookup) GetByID(
 ) (*token.Token, error) {
 	return m.svc.GetByID(ctx, id)
 }
-
-type mysqlEventRecorder struct {
-	repo   *event.Repository
-	tokens *token.Repository
-	logger *slog.Logger
-}
-
-func (m *mysqlEventRecorder) Record(
-	ctx context.Context,
-	t *token.Token,
-	evt *event.Event,
-) error {
-	if err := m.repo.Insert(ctx, evt); err != nil {
-		return fmt.Errorf("insert event: %w", err)
-	}
-	if err := m.tokens.IncrementTriggerCount(ctx, t.ID); err != nil {
-		m.logger.WarnContext(ctx, "mysql: increment trigger count",
-			"error", err, "token_id", t.ID)
-	}
-	return nil
-}

PROJECTS/beginner/canary-token-generator/backend/internal/config/config.go

@@ -26,6 +26,20 @@ type Config struct {
 	Canary    CanaryConfig    `koanf:"canary"`
 	Turnstile TurnstileConfig `koanf:"turnstile"`
 	MySQL     MySQLConfig     `koanf:"mysql"`
+	Notify    NotifyConfig    `koanf:"notify"`
+}
+
+type NotifyConfig struct {
+	DedupTTL          time.Duration `koanf:"dedup_ttl"`
+	SendTimeout       time.Duration `koanf:"send_timeout"`
+	MaxTries          uint          `koanf:"max_tries"`
+	MaxElapsed        time.Duration `koanf:"max_elapsed"`
+	InitialInterval   time.Duration `koanf:"initial_interval"`
+	RetentionInterval time.Duration `koanf:"retention_interval"`
+	RetentionLimit    int           `koanf:"retention_limit"`
+	WebhookHMACSecret string        `koanf:"webhook_hmac_secret"`
+	TelegramAPIBase   string        `koanf:"telegram_api_base"`
+	FingerprintWindow time.Duration `koanf:"fingerprint_window"`
 }
 
 type CanaryConfig struct {
@@ -75,9 +89,13 @@ type RedisConfig struct {
 }
 
 type RateLimitConfig struct {
-	Requests int           `koanf:"requests"`
-	Window   time.Duration `koanf:"window"`
-	Burst    int           `koanf:"burst"`
+	Requests        int           `koanf:"requests"`
+	Window          time.Duration `koanf:"window"`
+	Burst           int           `koanf:"burst"`
+	CreateMinRate   int           `koanf:"create_min_rate"`
+	CreateMinBurst  int           `koanf:"create_min_burst"`
+	CreateHourRate  int           `koanf:"create_hour_rate"`
+	CreateHourBurst int           `koanf:"create_hour_burst"`
 }
 
 type CORSConfig struct {
@@ -182,9 +200,13 @@ func loadDefaults(k *koanf.Koanf) error {
 		"redis.pool_size":      10,
 		"redis.min_idle_conns": 5,
 
-		"rate_limit.requests": 100,
-		"rate_limit.window":   "1m",
-		"rate_limit.burst":    20,
+		"rate_limit.requests":          100,
+		"rate_limit.window":            "1m",
+		"rate_limit.burst":             20,
+		"rate_limit.create_min_rate":   5,
+		"rate_limit.create_min_burst":  5,
+		"rate_limit.create_hour_rate":  20,
+		"rate_limit.create_hour_burst": 5,
 
 		"cors.allowed_origins": []string{"http://localhost:3000"},
 		"cors.allowed_methods": []string{
@@ -222,6 +244,17 @@ func loadDefaults(k *koanf.Koanf) error {
 		"mysql.addr":        "0.0.0.0:3306",
 		"mysql.public_host": "localhost",
 		"mysql.public_port": 3306,
+
+		"notify.dedup_ttl":           "15m",
+		"notify.send_timeout":        "30s",
+		"notify.max_tries":           3,
+		"notify.max_elapsed":         "30s",
+		"notify.initial_interval":    "500ms",
+		"notify.retention_interval":  "1h",
+		"notify.retention_limit":     100,
+		"notify.webhook_hmac_secret": "",
+		"notify.telegram_api_base":   "https://api.telegram.org",
+		"notify.fingerprint_window":  "5m",
 	}
 
 	for key, value := range defaults {
@@ -234,30 +267,44 @@ func loadDefaults(k *koanf.Koanf) error {
 }
 
 var envKeyMap = map[string]string{
-	"DATABASE_URL":                "database.url",
-	"REDIS_URL":                   "redis.url",
-	"APP_ENVIRONMENT":             "app.environment",
-	"HOST":                        "server.host",
-	"PORT":                        "server.port",
-	"LOG_LEVEL":                   "log.level",
-	"LOG_FORMAT":                  "log.format",
-	"RATE_LIMIT_REQUESTS":         "rate_limit.requests",
-	"RATE_LIMIT_WINDOW":           "rate_limit.window",
-	"RATE_LIMIT_BURST":            "rate_limit.burst",
-	"OTEL_ENDPOINT":               "otel.endpoint",
-	"OTEL_EXPORTER_OTLP_ENDPOINT": "otel.endpoint",
-	"OTEL_SERVICE_NAME":           "otel.service_name",
-	"OTEL_ENABLED":                "otel.enabled",
-	"OTEL_INSECURE":               "otel.insecure",
-	"OTEL_SAMPLE_RATE":            "otel.sample_rate",
-	"CANARY_BASE_URL":             "canary.base_url",
-	"CANARY_MANAGE_URL":           "canary.manage_url",
-	"TURNSTILE_SECRET_KEY":        "turnstile.secret_key",
-	"TURNSTILE_SITE_KEY":          "turnstile.site_key",
-	"MYSQL_ENABLED":               "mysql.enabled",
-	"MYSQL_ADDR":                  "mysql.addr",
-	"MYSQL_PUBLIC_HOST":           "mysql.public_host",
-	"MYSQL_PUBLIC_PORT":           "mysql.public_port",
+	"DATABASE_URL":                 "database.url",
+	"REDIS_URL":                    "redis.url",
+	"APP_ENVIRONMENT":              "app.environment",
+	"HOST":                         "server.host",
+	"PORT":                         "server.port",
+	"LOG_LEVEL":                    "log.level",
+	"LOG_FORMAT":                   "log.format",
+	"RATE_LIMIT_REQUESTS":          "rate_limit.requests",
+	"RATE_LIMIT_WINDOW":            "rate_limit.window",
+	"RATE_LIMIT_BURST":             "rate_limit.burst",
+	"OTEL_ENDPOINT":                "otel.endpoint",
+	"OTEL_EXPORTER_OTLP_ENDPOINT":  "otel.endpoint",
+	"OTEL_SERVICE_NAME":            "otel.service_name",
+	"OTEL_ENABLED":                 "otel.enabled",
+	"OTEL_INSECURE":                "otel.insecure",
+	"OTEL_SAMPLE_RATE":             "otel.sample_rate",
+	"CANARY_BASE_URL":              "canary.base_url",
+	"CANARY_MANAGE_URL":            "canary.manage_url",
+	"TURNSTILE_SECRET_KEY":         "turnstile.secret_key",
+	"TURNSTILE_SITE_KEY":           "turnstile.site_key",
+	"MYSQL_ENABLED":                "mysql.enabled",
+	"MYSQL_ADDR":                   "mysql.addr",
+	"MYSQL_PUBLIC_HOST":            "mysql.public_host",
+	"MYSQL_PUBLIC_PORT":            "mysql.public_port",
+	"RATE_LIMIT_CREATE_MIN_RATE":   "rate_limit.create_min_rate",
+	"RATE_LIMIT_CREATE_MIN_BURST":  "rate_limit.create_min_burst",
+	"RATE_LIMIT_CREATE_HOUR_RATE":  "rate_limit.create_hour_rate",
+	"RATE_LIMIT_CREATE_HOUR_BURST": "rate_limit.create_hour_burst",
+	"NOTIFY_DEDUP_TTL":             "notify.dedup_ttl",
+	"NOTIFY_SEND_TIMEOUT":          "notify.send_timeout",
+	"NOTIFY_MAX_TRIES":             "notify.max_tries",
+	"NOTIFY_MAX_ELAPSED":           "notify.max_elapsed",
+	"NOTIFY_INITIAL_INTERVAL":      "notify.initial_interval",
+	"NOTIFY_RETENTION_INTERVAL":    "notify.retention_interval",
+	"NOTIFY_RETENTION_LIMIT":       "notify.retention_limit",
+	"WEBHOOK_HMAC_SECRET":          "notify.webhook_hmac_secret",
+	"NOTIFY_TELEGRAM_API_BASE":     "notify.telegram_api_base",
+	"NOTIFY_FINGERPRINT_WINDOW":    "notify.fingerprint_window",
 }
 
 func envKeyReplacer(s string) string {

PROJECTS/beginner/canary-token-generator/backend/internal/token/dto.go

@@ -17,7 +17,7 @@ type CreateRequest struct {
 	TelegramChat  string          `json:"telegram_chat"         validate:"required_if=AlertChannel telegram"`
 	WebhookURL    string          `json:"webhook_url"           validate:"required_if=AlertChannel webhook,omitempty,url"`
 	Metadata      json.RawMessage `json:"metadata"`
-	TurnstileResp string          `json:"cf_turnstile_response" validate:"required"`
+	TurnstileResp string          `json:"cf_turnstile_response"`
 }
 
 type Response struct {