sqs: add the admin purge/peek audit line and counters (§3.6)

adapter/sqs.go

@@ -3,6 +3,7 @@ package adapter
 import (
 	"context"
 	"io"
+	"log/slog"
 	"net"
 	"net/http"
 	"strconv"
@@ -214,6 +215,16 @@ type SQSServer struct {
 	// nil on non-monitored fixtures; observeThrottleDecision is
 	// nil-safe so the request path pays one branch when unwired.
 	throttleObserver SQSThrottleObserver
+	// adminObserver records the §3.6 admin purge / peek counters.
+	// nil on non-monitored fixtures; the increment helpers are
+	// nil-safe so an unwired server pays one branch.
+	adminObserver SQSAdminObserver
+	// adminAuditLogger is where the §3.6 admin audit lines go. Defaults
+	// to slog.Default(); production installs the component="admin" child
+	// logger so these records land in the same destination, with the same
+	// attributes, as every other admin audit entry rather than bypassing
+	// it through the process-wide default.
+	adminAuditLogger *slog.Logger
 }
 
 // SQSPartitionObserver is the metrics-package interface
@@ -224,6 +235,14 @@ type SQSPartitionObserver interface {
 	ObservePartitionMessage(queue string, partition uint32, action string)
 }
 
+// SQSAdminObserver is the metrics-package interface
+// (monitoring.SQSMetrics) re-declared here so the adapter does not
+// import monitoring at the package boundary.
+type SQSAdminObserver interface {
+	ObserveAdminPurgeQueue(queue string, outcome string)
+	ObserveAdminPeekQueue(queue string, outcome string)
+}
+
 // SQSThrottleObserver is the metrics-package interface
 // (monitoring.SQSThrottleObserver) re-declared here so the adapter
 // does not import monitoring at the package boundary.
@@ -278,6 +297,25 @@ func WithSQSLeaderMap(m map[string]string) SQSServerOption {
 	}
 }
 
+// WithSQSAdminObserver installs the §3.6 admin purge / peek counters.
+func WithSQSAdminObserver(o SQSAdminObserver) SQSServerOption {
+	return func(s *SQSServer) {
+		if o != nil {
+			s.adminObserver = o
+		}
+	}
+}
+
+// WithSQSAdminAuditLogger routes the §3.6 admin audit lines to the
+// supplied logger instead of slog.Default(). No-ops on nil.
+func WithSQSAdminAuditLogger(l *slog.Logger) SQSServerOption {
+	return func(s *SQSServer) {
+		if l != nil {
+			s.adminAuditLogger = l
+		}
+	}
+}
+
 // WithSQSPartitionObserver installs the
 // elastickv_sqs_partition_messages_total counter observer on the
 // SQS server. Pass nil (the default) on non-monitored test

adapter/sqs_admin.go

@@ -3,6 +3,7 @@ package adapter
 import (
 	"bytes"
 	"context"
+	"log/slog"
 	"sort"
 	"strconv"
 	"strings"
@@ -313,19 +314,29 @@ func (e *PurgeInProgressError) Is(target error) bool {
 //   - ErrAdminSQSNotFound        — queue absent
 //   - *PurgeInProgressError      — last purge < 60 s ago; errors.Is matches
 //     ErrAdminSQSPurgeInProgress, RetryAfter carries the duration.
-//   - ErrAdminSQSValidation      — empty / whitespace name
+//   - ErrAdminSQSValidation      — malformed queue name
 func (s *SQSServer) AdminPurgeQueue(ctx context.Context, principal AdminPrincipal, name string) (AdminPurgeResult, error) {
 	if !principal.Role.canWrite() {
+		s.recordAdminPurge(ctx, principal, name, adminOutcomeForbidden)
 		return AdminPurgeResult{}, ErrAdminForbidden
 	}
 	if !isVerifiedSQSLeader(ctx, s.coordinator) {
+		s.recordAdminPurge(ctx, principal, name, adminOutcomeNotLeader)
 		return AdminPurgeResult{}, ErrAdminNotLeader
 	}
-	if strings.TrimSpace(name) == "" {
+	if strings.TrimSpace(name) != name || validateQueueName(name) != nil {
+		s.recordAdminPurge(ctx, principal, name, adminOutcomeValidation)
 		return AdminPurgeResult{}, ErrAdminSQSValidation
 	}
 	oldGen, newGen, err := s.purgeQueueWithRetry(ctx, name)
 	if err != nil {
+		// Every refusal gets the operation-specific audit record, not
+		// just the counter. A repeated purge inside the 60-second window
+		// used to return from here with no admin.sqs.purge_queue line at
+		// all, leaving only the generic HTTP audit middleware's status
+		// and path -- which cannot say WHY it was refused, and so cannot
+		// answer the question the §3.6 signal exists for.
+		s.recordAdminPurge(ctx, principal, name, adminPurgeOutcomeForError(err))
 		var rateLimit *purgeRateLimitedError
 		if errors.As(err, &rateLimit) {
 			return AdminPurgeResult{}, &PurgeInProgressError{RetryAfter: rateLimit.remaining}
@@ -335,9 +346,111 @@ func (s *SQSServer) AdminPurgeQueue(ctx context.Context, principal AdminPrincipa
 		}
 		return AdminPurgeResult{}, errors.Wrap(err, "admin purge queue")
 	}
+	s.recordAdminPurge(ctx, principal, name, adminOutcomeOK,
+		slog.Uint64("generation_before", oldGen),
+		slog.Uint64("generation_after", newGen))
 	return AdminPurgeResult{GenerationBefore: oldGen, GenerationAfter: newGen}, nil
 }
 
+// recordAdminPurge emits the §3.6 audit line and bumps the counter for one
+// purge outcome.
+//
+// Both in one place so an exit path cannot record the metric and skip the
+// audit, which is how the purge-in-progress refusal ended up counted but
+// never audited. Generation attributes are passed only by the success path:
+// a refusal has no committed generation pair, and inventing one would put a
+// state that never existed into the audit trail.
+func (s *SQSServer) recordAdminPurge(
+	ctx context.Context,
+	principal AdminPrincipal,
+	name string,
+	outcome string,
+	extra ...slog.Attr,
+) {
+	s.observeAdminPurge(name, outcome)
+	// access_key, role, queue, outcome.
+	const baseAuditAttrs = 4
+	attrs := make([]any, 0, len(extra)+baseAuditAttrs)
+	attrs = append(attrs,
+		// AdminPrincipal carries AccessKey, not the design's
+		// "subject": the access key ID is the identity the admin
+		// surface authenticates, and it is an identifier rather than
+		// a secret (the signing key never appears here).
+		slog.String("access_key", principal.AccessKey),
+		slog.String("role", string(principal.Role)),
+		slog.String("queue", name),
+		slog.String("outcome", outcome))
+	for _, attr := range extra {
+		attrs = append(attrs, attr)
+	}
+	s.adminLogger().InfoContext(ctx, "admin.sqs.purge_queue", attrs...)
+}
+
+// adminLogger returns the configured audit destination, falling back to the
+// process default so a server built without the option still audits.
+func (s *SQSServer) adminLogger() *slog.Logger {
+	if s == nil || s.adminAuditLogger == nil {
+		return slog.Default()
+	}
+	return s.adminAuditLogger
+}
+
+// Outcome labels for the §3.6 admin counters, mirrored from
+// monitoring so the adapter does not import it at this boundary.
+const (
+	adminOutcomeOK              = "ok"
+	adminOutcomeForbidden       = "forbidden"
+	adminOutcomeNotLeader       = "not_leader"
+	adminOutcomeNotFound        = "not_found"
+	adminOutcomeValidation      = "validation"
+	adminOutcomePurgeInProgress = "purge_in_progress"
+	// NOTE: no "throttled" here — the admin peek throttle is a
+	// separate deferred follow-up, and declaring the label before a
+	// call site exists would imply coverage the code does not have.
+	adminOutcomeInternalError = "internal_error"
+)
+
+// adminPurgeOutcomeForError classifies a purge failure by SENTINEL,
+// never by message text: an error-string label would let one recurring
+// failure grow the series set without bound.
+func adminPurgeOutcomeForError(err error) string {
+	var rateLimit *purgeRateLimitedError
+	switch {
+	case errors.As(err, &rateLimit):
+		return adminOutcomePurgeInProgress
+	case isSQSAdminQueueDoesNotExist(err):
+		return adminOutcomeNotFound
+	case errors.Is(err, ErrAdminNotLeader), isTransientLeaderRedisError(err):
+		return adminOutcomeNotLeader
+	default:
+		return adminOutcomeInternalError
+	}
+}
+
+// AdminObserver exposes the §3.6 counters so the admin HTTP handler can
+// record the rejections it serves before reaching this adapter. Returns nil
+// on a server built without an observer.
+func (s *SQSServer) AdminObserver() SQSAdminObserver {
+	if s == nil {
+		return nil
+	}
+	return s.adminObserver
+}
+
+func (s *SQSServer) observeAdminPurge(queue, outcome string) {
+	if s == nil || s.adminObserver == nil {
+		return
+	}
+	s.adminObserver.ObserveAdminPurgeQueue(queue, outcome)
+}
+
+func (s *SQSServer) observeAdminPeek(queue, outcome string) {
+	if s == nil || s.adminObserver == nil {
+		return
+	}
+	s.adminObserver.ObserveAdminPeekQueue(queue, outcome)
+}
+
 // AdminSetQueueAttributes is the SigV4-bypass counterpart to
 // SetQueueAttributes. It is intentionally generic rather than
 // DLQ-specific so the admin SPA can edit RedrivePolicy and