1
0
forked from baron/baron-sso

adminfront 개요 통계 추가

This commit is contained in:
2026-05-06 16:14:52 +09:00
parent 6cdd0fd81e
commit 13dee9ae9b
24 changed files with 2082 additions and 297 deletions

View File

@@ -183,11 +183,15 @@ func main() {
chDB := getEnv("CLICKHOUSE_DB", "baron_sso")
var auditRepo domain.AuditRepository
var rpUsageProjectionRepo domain.RPUsageProjectionRepository
var rpUsageQueryRepo domain.RPUsageQueryRepository
if repo, err := repository.NewClickHouseRepository(chHost, chPort, chUser, chPass, chDB); err != nil {
slog.Warn("Failed to connect to ClickHouse. Audit logs will fail.", "error", err)
auditRepo = nil // Explicitly set to nil interface
} else {
auditRepo = repo
rpUsageProjectionRepo = repo
rpUsageQueryRepo = repo
slog.Info("✅ Connected to ClickHouse")
}
@@ -297,6 +301,7 @@ func main() {
userGroupRepo := repository.NewUserGroupRepository(db)
userRepo := repository.NewUserRepository(db)
ketoOutboxRepo := repository.NewKetoOutboxRepository(db) // Reuse or re-init
rpUsageOutboxRepo := repository.NewRPUsageOutboxRepository(db)
worksmobileOutboxRepo := repository.NewWorksmobileOutboxRepository(db)
sharedLinkRepo := repository.NewSharedLinkRepository(db)
kratosAdminService := service.NewKratosAdminService()
@@ -323,6 +328,14 @@ func main() {
worksmobileRelayWorker := service.NewWorksmobileRelayWorker(worksmobileOutboxRepo, worksmobileClient)
go worksmobileRelayWorker.Start(context.Background())
slog.Info("✅ Worksmobile Relay Worker started")
rpUsageEmitter := service.NewRPUsageEventEmitter(rpUsageOutboxRepo)
if rpUsageProjectionRepo != nil {
rpUsageProjectorWorker := service.NewRPUsageProjectorWorker(rpUsageOutboxRepo, rpUsageProjectionRepo)
go rpUsageProjectorWorker.Start(context.Background())
slog.Info("✅ RP Usage Projector Worker started")
} else {
slog.Warn("RP Usage Projector Worker skipped because ClickHouse is unavailable")
}
sharedLinkService := service.NewSharedLinkService(sharedLinkRepo)
userGroupService := service.NewUserGroupService(userGroupRepo, userRepo, tenantRepo, ketoService, ketoOutboxRepo, kratosAdminService)
tenantService.SetKetoService(ketoService) // Keto 주입
@@ -342,7 +355,12 @@ func main() {
authHandler := handler.NewAuthHandler(redisService, idpProvider, auditRepo, oathkeeperRepo, tenantService, ketoService, ketoOutboxRepo, userRepo, consentRepo, kratosAdminService)
authHandler.HeadlessJWKS = headlessJWKSCache
authHandler.RPUserMetadataRepo = rpUserMetadataRepo
authHandler.RPUsageSink = rpUsageEmitter
adminHandler := handler.NewAdminHandler(ketoService, ketoOutboxRepo)
adminHandler.RPUsageQueries = rpUsageQueryRepo
adminHandler.TenantRepo = tenantRepo
adminHandler.Hydra = hydraService
adminHandler.AuditRepo = auditRepo
devHandler := handler.NewDevHandler(redisService, secretRepo, consentRepo, relyingPartyService, ketoService, ketoOutboxRepo, tenantService, developerService, authHandler)
devHandler.HeadlessJWKS = headlessJWKSCache
devHandler.AuditRepo = auditRepo
@@ -674,6 +692,7 @@ func main() {
admin.Get("/check", adminHandler.CheckAuth) // 기본 Admin 체크는 requireAdmin 없이 ApiKeyAuth로만 보호될 수 있음 (또는 추가 가능)
admin.Get("/stats", requireSuperAdmin, adminHandler.GetSystemStats)
admin.Get("/rp-usage/daily", requireAdmin, adminHandler.GetRPUsageDaily)
// Tenant Management (Mixed roles, handler filters results)
admin.Get("/tenants", requireAnyUser, tenantHandler.ListTenants)

View File

@@ -45,6 +45,7 @@ func migrateSchemas(db *gorm.DB) error {
&domain.ClientSecret{},
&domain.ClientConsent{},
&domain.KetoOutbox{},
&domain.RPUsageEvent{},
&domain.WorksmobileOutbox{},
&domain.WorksmobileResourceMapping{},
&domain.SharedLink{},

View File

@@ -26,6 +26,7 @@ type AuditRepository interface {
Create(log *AuditLog) error
FindPage(ctx context.Context, limit int, cursor *AuditCursor, tenantID string) ([]AuditLog, error)
FindByUserAndEvents(ctx context.Context, userID string, eventTypes []string, limit int) ([]AuditLog, error)
CountEventsSince(ctx context.Context, since time.Time) (int64, error)
CountFailuresSince(ctx context.Context, since time.Time, tenantID string) (int64, error)
CountActiveSessionsSince(ctx context.Context, since time.Time, tenantID string) (int64, error)
Ping(ctx context.Context) error

View File

@@ -0,0 +1,101 @@
package domain
import (
"context"
"time"
"github.com/google/uuid"
"github.com/lib/pq"
"gorm.io/gorm"
)
const (
RPUsageOutboxStatusPending = "pending"
RPUsageOutboxStatusProcessing = "processing"
RPUsageOutboxStatusProcessed = "processed"
RPUsageOutboxStatusFailed = "failed"
)
const (
RPUsageEventTypeAuthorizationGranted = "rp_usage.authorization_granted"
RPUsageEventTypeAuthorizationRevoked = "rp_usage.authorization_revoked"
)
const (
RPUsageTenantTypeCompany = TenantTypeCompany
RPUsageTenantTypeOrganization = TenantTypeOrganization
)
type RPUsageEvent struct {
ID string `gorm:"primaryKey;type:uuid;default:gen_random_uuid()" json:"id"`
EventType string `gorm:"not null;index:idx_rp_usage_outbox_event" json:"eventType"`
Subject string `gorm:"not null;index:idx_rp_usage_outbox_subject" json:"subject"`
TenantID string `gorm:"index:idx_rp_usage_outbox_tenant" json:"tenantId,omitempty"`
TenantType string `gorm:"index:idx_rp_usage_outbox_tenant" json:"tenantType,omitempty"`
ClientID string `gorm:"not null;index:idx_rp_usage_outbox_client" json:"clientId"`
ClientName string `json:"clientName,omitempty"`
SessionID string `gorm:"index" json:"sessionId,omitempty"`
Scopes pq.StringArray `gorm:"type:text[]" json:"scopes,omitempty"`
Source string `gorm:"not null;index" json:"source"`
CorrelationID string `gorm:"index" json:"correlationId,omitempty"`
Payload JSONMap `gorm:"type:jsonb" json:"payload,omitempty"`
DedupeKey string `gorm:"uniqueIndex" json:"dedupeKey"`
Status string `gorm:"default:'pending';index" json:"status"`
RetryCount int `gorm:"default:0" json:"retryCount"`
LastError string `json:"lastError,omitempty"`
NextAttemptAt *time.Time `json:"nextAttemptAt,omitempty"`
OccurredAt time.Time `gorm:"not null;index" json:"occurredAt"`
ProcessedAt *time.Time `json:"processedAt,omitempty"`
CreatedAt time.Time `json:"createdAt"`
UpdatedAt time.Time `json:"updatedAt"`
}
func (e *RPUsageEvent) TableName() string {
return "rp_usage_outbox"
}
func (e *RPUsageEvent) BeforeCreate(tx *gorm.DB) error {
if e.ID == "" {
e.ID = uuid.NewString()
}
if e.Status == "" {
e.Status = RPUsageOutboxStatusPending
}
if e.OccurredAt.IsZero() {
e.OccurredAt = time.Now()
}
if e.Payload == nil {
e.Payload = JSONMap{}
}
return nil
}
type RPUsageEventSink interface {
EmitRPUsageEvent(ctx context.Context, event RPUsageEvent) error
}
type RPUsageProjectionRepository interface {
CreateRPUsageEvent(ctx context.Context, event RPUsageEvent) error
}
type RPUsageDailyMetric struct {
Date string `json:"date"`
TenantID string `json:"tenantId"`
TenantType string `json:"tenantType"`
TenantName string `json:"tenantName,omitempty"`
ClientID string `json:"clientId"`
ClientName string `json:"clientName"`
LoginRequests uint64 `json:"loginRequests"`
OtherRequests uint64 `json:"otherRequests"`
UniqueSubjects uint64 `json:"uniqueSubjects"`
}
type RPUsageQuery struct {
Days int
Period string
TenantID string
}
type RPUsageQueryRepository interface {
FindRPUsage(ctx context.Context, query RPUsageQuery) ([]RPUsageDailyMetric, error)
}

View File

@@ -1,17 +1,29 @@
package handler
import (
"baron-sso-backend/internal/domain"
"baron-sso-backend/internal/repository"
"baron-sso-backend/internal/service"
"context"
"runtime"
"strconv"
"strings"
"time"
"github.com/gofiber/fiber/v2"
)
type adminHydraClientLister interface {
ListClients(ctx context.Context, limit, offset int) ([]domain.HydraClient, error)
}
type AdminHandler struct {
Keto service.KetoService
KetoOutbox repository.KetoOutboxRepository
Keto service.KetoService
KetoOutbox repository.KetoOutboxRepository
RPUsageQueries domain.RPUsageQueryRepository
TenantRepo repository.TenantRepository
Hydra adminHydraClientLister
AuditRepo domain.AuditRepository
}
func NewAdminHandler(keto service.KetoService, ketoOutbox repository.KetoOutboxRepository) *AdminHandler {
@@ -21,6 +33,76 @@ func NewAdminHandler(keto service.KetoService, ketoOutbox repository.KetoOutboxR
}
}
func (h *AdminHandler) GetRPUsageDaily(c *fiber.Ctx) error {
if h == nil || h.RPUsageQueries == nil {
return c.Status(fiber.StatusServiceUnavailable).JSON(fiber.Map{
"error": "rp usage query service unavailable",
})
}
days := 14
if raw := c.Query("days"); raw != "" {
if parsed, err := strconv.Atoi(raw); err == nil {
days = parsed
}
}
period := normalizeRPUsagePeriod(c.Query("period"))
tenantID, allowed := h.authorizedRPUsageTenantID(c, strings.TrimSpace(c.Query("tenantId")))
if !allowed {
return c.Status(fiber.StatusForbidden).JSON(fiber.Map{
"error": "forbidden: tenant rp usage stats permission denied",
})
}
items, err := h.RPUsageQueries.FindRPUsage(c.Context(), domain.RPUsageQuery{
Days: days,
Period: period,
TenantID: tenantID,
})
if err != nil {
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{
"error": err.Error(),
})
}
return c.JSON(fiber.Map{
"items": items,
"days": days,
"period": period,
"tenantId": tenantID,
})
}
func normalizeRPUsagePeriod(period string) string {
switch strings.ToLower(strings.TrimSpace(period)) {
case "week":
return "week"
case "month":
return "month"
default:
return "day"
}
}
func (h *AdminHandler) authorizedRPUsageTenantID(c *fiber.Ctx, requestedTenantID string) (string, bool) {
profile, _ := c.Locals("user_profile").(*domain.UserProfileResponse)
if profile != nil && domain.NormalizeRole(profile.Role) == domain.RoleSuperAdmin {
return requestedTenantID, true
}
tenantID := requestedTenantID
if tenantID == "" && profile != nil && profile.TenantID != nil {
tenantID = strings.TrimSpace(*profile.TenantID)
}
if tenantID == "" {
return "", false
}
if h == nil || h.Keto == nil || profile == nil || strings.TrimSpace(profile.ID) == "" {
return "", false
}
allowed, err := h.Keto.CheckPermission(c.Context(), "User:"+profile.ID, "Tenant", tenantID, "view_rp_usage_stats")
if err != nil || !allowed {
return "", false
}
return tenantID, true
}
func (h *AdminHandler) CheckAuth(c *fiber.Ctx) error {
return c.Status(fiber.StatusOK).JSON(fiber.Map{"status": "ok"})
}
@@ -29,10 +111,14 @@ func (h *AdminHandler) CheckAuth(c *fiber.Ctx) error {
func (h *AdminHandler) GetSystemStats(c *fiber.Ctx) error {
var m runtime.MemStats
runtime.ReadMemStats(&m)
ctx := c.Context()
stats := fiber.Map{
"goroutines": runtime.NumGoroutine(),
"cpus": runtime.NumCPU(),
"totalTenants": h.countTenants(ctx),
"oidcClients": h.countOIDCClients(ctx),
"auditEvents24h": h.countAuditEventsSince(ctx, time.Now().UTC().Add(-24*time.Hour)),
"goroutines": runtime.NumGoroutine(),
"cpus": runtime.NumCPU(),
"memory": fiber.Map{
"alloc": m.Alloc,
"totalAlign": m.TotalAlloc,
@@ -44,3 +130,59 @@ func (h *AdminHandler) GetSystemStats(c *fiber.Ctx) error {
return c.Status(fiber.StatusOK).JSON(stats)
}
func (h *AdminHandler) countTenants(ctx context.Context) int64 {
if h == nil || h.TenantRepo == nil {
return 0
}
_, total, err := h.TenantRepo.List(ctx, 1, 0, "")
if err != nil {
return 0
}
return total
}
func (h *AdminHandler) countOIDCClients(ctx context.Context) int64 {
if h == nil || h.Hydra == nil {
return 0
}
const pageSize = 500
var total int64
for offset := 0; ; offset += pageSize {
clients, err := h.Hydra.ListClients(ctx, pageSize, offset)
if err != nil {
return total
}
for _, client := range clients {
if isHiddenSystemClient(client) {
continue
}
total++
}
if len(clients) < pageSize {
break
}
}
return total
}
func (h *AdminHandler) countAuditEventsSince(ctx context.Context, since time.Time) int64 {
if h == nil || h.AuditRepo == nil {
return 0
}
count, err := h.AuditRepo.CountEventsSince(ctx, since)
if err == nil && count > 0 {
return count
}
logs, pageErr := h.AuditRepo.FindPage(ctx, 10000, nil, "")
if pageErr != nil {
return count
}
var fallbackCount int64
for _, log := range logs {
if !log.Timestamp.Before(since) {
fallbackCount++
}
}
return fallbackCount
}

View File

@@ -0,0 +1,156 @@
package handler
import (
"baron-sso-backend/internal/domain"
"baron-sso-backend/internal/service"
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/gofiber/fiber/v2"
"github.com/stretchr/testify/require"
)
type fakeRPUsageQueryRepo struct {
query domain.RPUsageQuery
items []domain.RPUsageDailyMetric
}
func (f *fakeRPUsageQueryRepo) FindRPUsage(ctx context.Context, query domain.RPUsageQuery) ([]domain.RPUsageDailyMetric, error) {
f.query = query
return f.items, nil
}
type fakeAdminKeto struct {
allowed bool
subject string
object string
relation string
}
func (f *fakeAdminKeto) CheckPermission(ctx context.Context, subject, namespace, object, relation string) (bool, error) {
f.subject = subject
f.object = object
f.relation = relation
return f.allowed, nil
}
func (f *fakeAdminKeto) CreateRelation(ctx context.Context, namespace, object, relation, subject string) error {
return nil
}
func (f *fakeAdminKeto) DeleteRelation(ctx context.Context, namespace, object, relation, subject string) error {
return nil
}
func (f *fakeAdminKeto) ListRelations(ctx context.Context, namespace, object, relation, subject string) ([]service.RelationTuple, error) {
return nil, nil
}
func (f *fakeAdminKeto) ListObjects(ctx context.Context, namespace, relation, subject string) ([]string, error) {
return nil, nil
}
type fakeOverviewAuditRepo struct {
mockAuditRepo
since time.Time
count int64
}
func (f *fakeOverviewAuditRepo) CountEventsSince(ctx context.Context, since time.Time) (int64, error) {
f.since = since
return f.count, nil
}
func TestAdminHandler_GetRPUsageDaily(t *testing.T) {
repo := &fakeRPUsageQueryRepo{
items: []domain.RPUsageDailyMetric{
{
Date: "2026-05-06",
TenantID: "tenant-1",
TenantType: domain.TenantTypeCompany,
ClientID: "orgfront",
ClientName: "OrgFront",
LoginRequests: 12,
OtherRequests: 4,
UniqueSubjects: 8,
},
},
}
h := &AdminHandler{RPUsageQueries: repo}
app := fiber.New()
app.Use(func(c *fiber.Ctx) error {
c.Locals("user_profile", &domain.UserProfileResponse{Role: domain.RoleSuperAdmin})
return c.Next()
})
app.Get("/api/v1/admin/rp-usage/daily", h.GetRPUsageDaily)
req := httptest.NewRequest(http.MethodGet, "/api/v1/admin/rp-usage/daily?days=7&period=week&tenantId=tenant-1", nil)
resp, err := app.Test(req)
require.NoError(t, err)
require.Equal(t, http.StatusOK, resp.StatusCode)
require.Equal(t, 7, repo.query.Days)
require.Equal(t, "week", repo.query.Period)
require.Equal(t, "tenant-1", repo.query.TenantID)
var body struct {
Items []domain.RPUsageDailyMetric `json:"items"`
Days int `json:"days"`
Period string `json:"period"`
TenantID string `json:"tenantId"`
}
require.NoError(t, json.NewDecoder(resp.Body).Decode(&body))
require.Equal(t, 7, body.Days)
require.Equal(t, "week", body.Period)
require.Equal(t, "tenant-1", body.TenantID)
require.Len(t, body.Items, 1)
require.Equal(t, "orgfront", body.Items[0].ClientID)
require.Equal(t, uint64(12), body.Items[0].LoginRequests)
}
func TestAdminHandler_GetRPUsageDailyChecksTenantPermission(t *testing.T) {
repo := &fakeRPUsageQueryRepo{}
keto := &fakeAdminKeto{allowed: true}
h := &AdminHandler{RPUsageQueries: repo, Keto: keto}
app := fiber.New()
app.Use(func(c *fiber.Ctx) error {
c.Locals("user_profile", &domain.UserProfileResponse{
ID: "user-1",
Role: domain.RoleTenantAdmin,
})
return c.Next()
})
app.Get("/api/v1/admin/rp-usage/daily", h.GetRPUsageDaily)
req := httptest.NewRequest(http.MethodGet, "/api/v1/admin/rp-usage/daily?tenantId=tenant-allowed", nil)
resp, err := app.Test(req)
require.NoError(t, err)
require.Equal(t, http.StatusOK, resp.StatusCode)
require.Equal(t, "User:user-1", keto.subject)
require.Equal(t, "tenant-allowed", keto.object)
require.Equal(t, "view_rp_usage_stats", keto.relation)
require.Equal(t, "tenant-allowed", repo.query.TenantID)
}
func TestAdminHandler_GetSystemStatsIncludesOverviewMetrics(t *testing.T) {
auditRepo := &fakeOverviewAuditRepo{count: 22}
h := &AdminHandler{AuditRepo: auditRepo}
app := fiber.New()
app.Get("/api/v1/admin/stats", h.GetSystemStats)
req := httptest.NewRequest(http.MethodGet, "/api/v1/admin/stats", nil)
resp, err := app.Test(req)
require.NoError(t, err)
require.Equal(t, http.StatusOK, resp.StatusCode)
var body map[string]any
require.NoError(t, json.NewDecoder(resp.Body).Decode(&body))
require.Contains(t, body, "totalTenants")
require.Contains(t, body, "oidcClients")
require.Contains(t, body, "auditEvents24h")
require.Equal(t, float64(22), body["auditEvents24h"])
require.Equal(t, time.UTC, auditRepo.since.Location())
}

View File

@@ -29,6 +29,7 @@ import (
"github.com/go-jose/go-jose/v4"
josejwt "github.com/go-jose/go-jose/v4/jwt"
"github.com/gofiber/fiber/v2"
"github.com/lib/pq"
)
const (
@@ -101,6 +102,7 @@ type AuthHandler struct {
UserRepo repository.UserRepository
ConsentRepo repository.ClientConsentRepository
RPUserMetadataRepo repository.RPUserMetadataRepository
RPUsageSink domain.RPUsageEventSink
}
type signupState struct {
@@ -245,6 +247,92 @@ func NewAuthHandler(redisService domain.RedisRepository, idpProvider domain.Iden
}
}
func (h *AuthHandler) emitRPUsageAuthorizationGranted(c *fiber.Ctx, consentRequest *domain.HydraConsentRequest, profile *domain.UserProfileResponse, sessionID string, autoAccepted bool, correlationID string) error {
if consentRequest == nil {
return nil
}
return h.emitRPUsageEvent(c, domain.RPUsageEventTypeAuthorizationGranted, consentRequest.Subject, consentRequest.Client, consentRequest.RequestedScope, profile, sessionID, "hydra_consent", correlationID, domain.JSONMap{
"auto_accepted": autoAccepted,
"scopes": consentRequest.RequestedScope,
})
}
func (h *AuthHandler) emitRPUsageAuthorizationRevoked(c *fiber.Ctx, subject string, clientID string, profile *domain.UserProfileResponse, sessionID string) error {
return h.emitRPUsageEvent(c, domain.RPUsageEventTypeAuthorizationRevoked, subject, domain.HydraClient{ClientID: clientID}, nil, profile, sessionID, "hydra_consent", clientID, domain.JSONMap{})
}
func (h *AuthHandler) emitRPUsageEvent(c *fiber.Ctx, eventType string, subject string, client domain.HydraClient, scopes []string, profile *domain.UserProfileResponse, sessionID string, source string, correlationID string, payload domain.JSONMap) error {
if h.RPUsageSink == nil {
return nil
}
clientID := strings.TrimSpace(client.ClientID)
if clientID == "" || strings.TrimSpace(subject) == "" {
return nil
}
tenantID, tenantType := rpUsageTenantFromProfile(profile)
event := domain.RPUsageEvent{
EventType: eventType,
Subject: strings.TrimSpace(subject),
TenantID: tenantID,
TenantType: tenantType,
ClientID: clientID,
ClientName: strings.TrimSpace(client.ClientName),
SessionID: strings.TrimSpace(sessionID),
Scopes: pq.StringArray(scopes),
Source: source,
CorrelationID: strings.TrimSpace(correlationID),
Payload: payload,
OccurredAt: time.Now(),
}
if event.Payload == nil {
event.Payload = domain.JSONMap{}
}
if event.ClientName != "" {
event.Payload["client_name"] = event.ClientName
}
if tenantID != "" {
event.Payload["tenant_id"] = tenantID
}
if tenantType != "" {
event.Payload["tenant_type"] = tenantType
}
if c != nil {
event.Payload["ip_address"] = c.IP()
event.Payload["user_agent"] = string(c.Request().Header.UserAgent())
}
ctx := context.Background()
if c != nil && c.UserContext() != nil {
ctx = c.UserContext()
}
return h.RPUsageSink.EmitRPUsageEvent(ctx, event)
}
func rpUsageTenantFromProfile(profile *domain.UserProfileResponse) (string, string) {
if profile == nil {
return "", ""
}
tenantID := ""
if profile.SessionTenantID != nil {
tenantID = strings.TrimSpace(*profile.SessionTenantID)
}
if tenantID == "" && profile.TenantID != nil {
tenantID = strings.TrimSpace(*profile.TenantID)
}
tenantType := ""
if profile.Tenant != nil {
switch strings.ToUpper(strings.TrimSpace(profile.Tenant.Type)) {
case domain.TenantTypeCompany, domain.TenantTypeOrganization:
tenantType = strings.ToUpper(strings.TrimSpace(profile.Tenant.Type))
if tenantID == "" {
tenantID = strings.TrimSpace(profile.Tenant.ID)
}
case domain.TenantTypeUserGroup, domain.TenantTypePersonal:
return "", ""
}
}
return tenantID, tenantType
}
// --- Signup Flow Handlers ---
// CheckEmail - 이메일 사용 가능 여부를 확인합니다.
@@ -5323,6 +5411,12 @@ func (h *AuthHandler) RevokeLinkedRp(c *fiber.Ctx) error {
if err != nil || subject == "" {
return fiber.NewError(fiber.StatusUnauthorized, "Authentication required")
}
profile, profileErr := h.resolveCurrentProfile(c)
if (profileErr != nil || profile == nil) && subject != "" {
if fallbackProfile, fallbackErr := h.resolveProfileForSubject(c.Context(), subject); fallbackErr == nil {
profile = fallbackProfile
}
}
slog.Info("RevokeLinkedRp called", "subject", subject, "client_id", clientID)
@@ -5354,6 +5448,11 @@ func (h *AuthHandler) RevokeLinkedRp(c *fiber.Ctx) error {
})
}
if err := h.emitRPUsageAuthorizationRevoked(c, subject, clientID, profile, h.resolveCurrentSessionID(c)); err != nil {
slog.Error("failed to emit rp usage event for revoked consent", "error", err, "client_id", clientID, "subject", subject)
return fiber.NewError(fiber.StatusInternalServerError, "Failed to record RP usage event")
}
h.triggerBackchannelLogoutForClient(c.Context(), c, subject, clientID, "")
return c.Status(fiber.StatusOK).JSON(fiber.Map{
@@ -5434,6 +5533,10 @@ func (h *AuthHandler) GetConsentRequest(c *fiber.Ctx) error {
sessionClaims = h.withRPProfileClaims(c.Context(), sessionClaims, consentRequest.Client, consentRequest.Subject)
acceptResp, err := h.Hydra.AcceptConsentRequest(c.Context(), challenge, consentRequest, sessionClaims)
if err == nil {
if err := h.emitRPUsageAuthorizationGranted(c, consentRequest, profile, currentSessionID, true, challenge); err != nil {
slog.Error("failed to emit rp usage event for local consent auto-accept", "error", err, "client_id", consentRequest.Client.ClientID, "subject", consentRequest.Subject)
return fiber.NewError(fiber.StatusInternalServerError, "Failed to record RP usage event")
}
return c.JSON(acceptResp)
}
slog.Error("failed to force auto-accept based on local DB", "error", err)
@@ -5516,6 +5619,11 @@ func (h *AuthHandler) GetConsentRequest(c *fiber.Ctx) error {
})
}
if err := h.emitRPUsageAuthorizationGranted(c, consentRequest, profile, currentSessionID, true, challenge); err != nil {
slog.Error("failed to emit rp usage event for skip consent", "error", err, "client_id", consentRequest.Client.ClientID, "subject", consentRequest.Subject)
return fiber.NewError(fiber.StatusInternalServerError, "Failed to record RP usage event")
}
slog.Info("Consent skipped and auto-accepted", "subject", consentRequest.Subject, "client", consentRequest.Client.ClientID, "session_id", currentSessionID)
return c.JSON(acceptResp)
}
@@ -5705,6 +5813,11 @@ func (h *AuthHandler) AcceptConsentRequest(c *fiber.Ctx) error {
})
}
if err := h.emitRPUsageAuthorizationGranted(c, consentRequest, profile, currentSessionID, false, req.ConsentChallenge); err != nil {
slog.Error("failed to emit rp usage event for accepted consent", "error", err, "client_id", consentRequest.Client.ClientID, "subject", consentRequest.Subject)
return fiber.NewError(fiber.StatusInternalServerError, "Failed to record RP usage event")
}
return c.JSON(acceptResp)
}

View File

@@ -3,6 +3,7 @@ package handler
import (
"baron-sso-backend/internal/domain"
"baron-sso-backend/internal/service"
"baron-sso-backend/internal/utils"
"encoding/json"
"io"
"net/http"
@@ -38,12 +39,14 @@ func TestRevokeLinkedRp_Success(t *testing.T) {
defer func() { http.DefaultClient = origDefault }()
auditRepo := &mockAuditRepo{}
rpUsageSink := &mockRPUsageEventSink{}
h := &AuthHandler{
Hydra: &service.HydraAdminService{
AdminURL: "http://hydra.test",
HTTPClient: client,
},
AuditRepo: auditRepo,
AuditRepo: auditRepo,
RPUsageSink: rpUsageSink,
}
app := fiber.New()
app.Delete("/api/v1/user/rp/linked/:id", h.RevokeLinkedRp)
@@ -54,6 +57,16 @@ func TestRevokeLinkedRp_Success(t *testing.T) {
resp, _ := app.Test(req, -1)
assert.Equal(t, http.StatusOK, resp.StatusCode)
assert.Equal(t, 1, len(auditRepo.logs))
assert.Equal(t, "consent.revoked", auditRepo.logs[0].EventType)
assert.Equal(t, "user-123", auditRepo.logs[0].UserID)
assert.Equal(t, "success", auditRepo.logs[0].Status)
auditDetails, err := utils.ParseAuditDetails(auditRepo.logs[0].Details)
assert.NoError(t, err)
assert.Equal(t, "app-1", auditDetails["client_id"])
assert.Equal(t, 1, len(rpUsageSink.events))
assert.Equal(t, domain.RPUsageEventTypeAuthorizationRevoked, rpUsageSink.events[0].EventType)
assert.Equal(t, "user-123", rpUsageSink.events[0].Subject)
assert.Equal(t, "app-1", rpUsageSink.events[0].ClientID)
}
func TestRevokeLinkedRp_SendsBackchannelLogoutTokenWhenConfigured(t *testing.T) {

View File

@@ -3,6 +3,7 @@ package handler
import (
"baron-sso-backend/internal/domain"
"baron-sso-backend/internal/service"
"baron-sso-backend/internal/utils"
"bytes"
"context"
"encoding/json"
@@ -305,6 +306,7 @@ func TestGetConsentRequest_Skip_AutoAccept(t *testing.T) {
defer func() { http.DefaultClient = origDefault }()
consentRepo := &mockConsentRepo{}
rpUsageSink := &mockRPUsageEventSink{}
mockKratosAdmin := &MockKratosAdminServiceForConsent{}
h := &AuthHandler{
@@ -314,6 +316,7 @@ func TestGetConsentRequest_Skip_AutoAccept(t *testing.T) {
},
KratosAdmin: mockKratosAdmin,
ConsentRepo: consentRepo,
RPUsageSink: rpUsageSink,
}
mockKratosAdmin.On("GetIdentity", mock.Anything, "user-123").Return(&service.KratosIdentity{
ID: "user-123",
@@ -332,6 +335,11 @@ func TestGetConsentRequest_Skip_AutoAccept(t *testing.T) {
var body map[string]interface{}
json.NewDecoder(resp.Body).Decode(&body)
assert.Equal(t, "http://rp/cb", body["redirectTo"])
assert.Equal(t, 1, len(rpUsageSink.events))
assert.Equal(t, domain.RPUsageEventTypeAuthorizationGranted, rpUsageSink.events[0].EventType)
assert.Equal(t, "client-app", rpUsageSink.events[0].ClientID)
assert.Equal(t, "challenge-skip", rpUsageSink.events[0].CorrelationID)
assert.Equal(t, true, rpUsageSink.events[0].Payload["auto_accepted"])
}
func TestAcceptConsentRequest_Normal(t *testing.T) {
@@ -370,6 +378,7 @@ func TestAcceptConsentRequest_Normal(t *testing.T) {
auditRepo := &mockAuditRepo{}
consentRepo := &mockConsentRepo{}
rpUsageSink := &mockRPUsageEventSink{}
mockKratosAdmin := &MockKratosAdminServiceForConsent{}
h := &AuthHandler{
@@ -380,6 +389,7 @@ func TestAcceptConsentRequest_Normal(t *testing.T) {
KratosAdmin: mockKratosAdmin,
AuditRepo: auditRepo,
ConsentRepo: consentRepo,
RPUsageSink: rpUsageSink,
}
mockKratosAdmin.On("GetIdentity", mock.Anything, "user-123").Return(&service.KratosIdentity{
ID: "user-123",
@@ -402,6 +412,21 @@ func TestAcceptConsentRequest_Normal(t *testing.T) {
assert.Equal(t, http.StatusOK, resp.StatusCode)
assert.Equal(t, 1, len(auditRepo.logs))
assert.Equal(t, "consent.granted", auditRepo.logs[0].EventType)
assert.Equal(t, "user-123", auditRepo.logs[0].UserID)
assert.Equal(t, "success", auditRepo.logs[0].Status)
auditDetails, err := utils.ParseAuditDetails(auditRepo.logs[0].Details)
assert.NoError(t, err)
assert.Equal(t, "client-app", auditDetails["client_id"])
assert.Equal(t, "Test App", auditDetails["client_name"])
assert.Equal(t, []interface{}{"openid"}, auditDetails["scopes"])
assert.Equal(t, 1, len(rpUsageSink.events))
assert.Equal(t, domain.RPUsageEventTypeAuthorizationGranted, rpUsageSink.events[0].EventType)
assert.Equal(t, "user-123", rpUsageSink.events[0].Subject)
assert.Equal(t, "client-app", rpUsageSink.events[0].ClientID)
assert.Equal(t, "Test App", rpUsageSink.events[0].ClientName)
assert.Equal(t, []string{"openid"}, []string(rpUsageSink.events[0].Scopes))
assert.Equal(t, "hydra_consent", rpUsageSink.events[0].Source)
}
func TestAcceptConsentRequest_EnforcesMandatoryTenantScope(t *testing.T) {

View File

@@ -109,12 +109,29 @@ func (m *mockAuditRepo) CountFailuresSince(ctx context.Context, since time.Time,
return 0, nil
}
func (m *mockAuditRepo) CountEventsSince(ctx context.Context, since time.Time) (int64, error) {
return 0, nil
}
func (m *mockAuditRepo) CountActiveSessionsSince(ctx context.Context, since time.Time, tenantID string) (int64, error) {
return 0, nil
}
func (m *mockAuditRepo) Ping(ctx context.Context) error { return nil }
type mockRPUsageEventSink struct {
events []domain.RPUsageEvent
err error
}
func (m *mockRPUsageEventSink) EmitRPUsageEvent(ctx context.Context, event domain.RPUsageEvent) error {
if m.err != nil {
return m.err
}
m.events = append(m.events, event)
return nil
}
type mockOathkeeperRepo struct {
logs []domain.OathkeeperAccessLog
}

View File

@@ -40,6 +40,10 @@ func (m *MockAuditRepository) CountFailuresSince(ctx context.Context, since time
return 0, nil
}
func (m *MockAuditRepository) CountEventsSince(ctx context.Context, since time.Time) (int64, error) {
return 0, nil
}
func (m *MockAuditRepository) CountActiveSessionsSince(ctx context.Context, since time.Time, tenantID string) (int64, error) {
return 0, nil
}
@@ -73,6 +77,10 @@ func (r *recordingAuditRepository) CountFailuresSince(ctx context.Context, since
return 0, nil
}
func (r *recordingAuditRepository) CountEventsSince(ctx context.Context, since time.Time) (int64, error) {
return 0, nil
}
func (r *recordingAuditRepository) CountActiveSessionsSince(ctx context.Context, since time.Time, tenantID string) (int64, error) {
return 0, nil
}

View File

@@ -3,6 +3,7 @@ package repository
import (
"baron-sso-backend/internal/domain"
"context"
"encoding/json"
"fmt"
"time"
@@ -77,9 +78,73 @@ func NewClickHouseRepository(host string, port int, user, password, db string) (
return nil, fmt.Errorf("failed to alter table: %w", err)
}
if err := ensureRPUsageTables(context.Background(), conn); err != nil {
return nil, fmt.Errorf("failed to create rp usage tables: %w", err)
}
return &ClickHouseRepository{conn: conn}, nil
}
func ensureRPUsageTables(ctx context.Context, conn driver.Conn) error {
factQuery := `
CREATE TABLE IF NOT EXISTS rp_usage_events (
event_id String,
occurred_at DateTime64(3) DEFAULT now64(3),
event_type String,
subject String,
tenant_id String,
tenant_type String,
client_id String,
client_name String,
session_id String,
scopes Array(String),
source String,
correlation_id String,
payload String
) ENGINE = MergeTree()
ORDER BY (occurred_at, event_id)
`
if err := conn.Exec(ctx, factQuery); err != nil {
return err
}
aggregateQuery := `
CREATE TABLE IF NOT EXISTS rp_usage_daily_aggregate (
event_date Date,
tenant_id String,
tenant_type String,
client_id String,
client_name String,
event_type String,
events_count AggregateFunction(count),
unique_subjects AggregateFunction(uniqExact, String)
) ENGINE = AggregatingMergeTree()
ORDER BY (event_date, tenant_id, client_id, event_type)
`
if err := conn.Exec(ctx, aggregateQuery); err != nil {
return err
}
viewQuery := `
CREATE MATERIALIZED VIEW IF NOT EXISTS rp_usage_daily_aggregate_mv
TO rp_usage_daily_aggregate
AS
SELECT
toDate(occurred_at) AS event_date,
tenant_id,
tenant_type,
client_id,
any(client_name) AS client_name,
event_type,
countState() AS events_count,
uniqExactState(subject) AS unique_subjects
FROM rp_usage_events
WHERE tenant_type IN ('COMPANY', 'ORGANIZATION')
GROUP BY event_date, tenant_id, tenant_type, client_id, event_type
`
return conn.Exec(ctx, viewQuery)
}
func (r *ClickHouseRepository) Create(log *domain.AuditLog) error {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
@@ -106,6 +171,125 @@ func (r *ClickHouseRepository) Create(log *domain.AuditLog) error {
)
}
func (r *ClickHouseRepository) CreateRPUsageEvent(ctx context.Context, event domain.RPUsageEvent) error {
if r == nil || r.conn == nil {
return fmt.Errorf("clickhouse connection is nil")
}
if event.OccurredAt.IsZero() {
event.OccurredAt = time.Now()
}
payloadBytes, err := json.Marshal(event.Payload)
if err != nil {
return fmt.Errorf("failed to marshal rp usage payload: %w", err)
}
query := `
INSERT INTO rp_usage_events (
event_id, occurred_at, event_type, subject, tenant_id, tenant_type,
client_id, client_name, session_id, scopes, source, correlation_id, payload
)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
`
return r.conn.Exec(ctx, query,
event.ID,
event.OccurredAt,
event.EventType,
event.Subject,
event.TenantID,
event.TenantType,
event.ClientID,
event.ClientName,
event.SessionID,
[]string(event.Scopes),
event.Source,
event.CorrelationID,
string(payloadBytes),
)
}
func (r *ClickHouseRepository) FindRPUsage(ctx context.Context, rpQuery domain.RPUsageQuery) ([]domain.RPUsageDailyMetric, error) {
if r == nil || r.conn == nil {
return nil, fmt.Errorf("clickhouse connection is nil")
}
days := rpQuery.Days
if days <= 0 || days > 90 {
days = 14
}
periodExpr := "event_date"
switch rpQuery.Period {
case "week":
periodExpr = "toMonday(event_date)"
case "month":
periodExpr = "toStartOfMonth(event_date)"
case "day", "":
periodExpr = "event_date"
default:
periodExpr = "event_date"
}
query := fmt.Sprintf(`
SELECT
date,
tenant_id,
tenant_type,
client_id,
any(client_name) AS client_name,
sumIf(events, event_type = ?) AS login_requests,
sumIf(events, event_type != ?) AS other_requests,
max(unique_subjects) AS unique_subjects
FROM (
SELECT
toString(%s) AS date,
tenant_id,
tenant_type,
client_id,
any(client_name) AS client_name,
event_type,
countMerge(events_count) AS events,
uniqExactMerge(unique_subjects) AS unique_subjects
FROM rp_usage_daily_aggregate
WHERE event_date >= today() - ?
AND tenant_type IN ('COMPANY', 'ORGANIZATION')
`, periodExpr)
args := []any{domain.RPUsageEventTypeAuthorizationGranted, domain.RPUsageEventTypeAuthorizationGranted, days - 1}
if rpQuery.TenantID != "" {
query += " AND tenant_id = ?\n"
args = append(args, rpQuery.TenantID)
}
query += fmt.Sprintf(`
GROUP BY %s, tenant_id, tenant_type, client_id, event_type
)
GROUP BY date, tenant_id, tenant_type, client_id
ORDER BY date ASC, tenant_id ASC, client_id ASC
`, periodExpr)
rows, err := r.conn.Query(ctx, query, args...)
if err != nil {
return nil, fmt.Errorf("failed to query rp usage daily aggregate: %w", err)
}
defer rows.Close()
metrics := make([]domain.RPUsageDailyMetric, 0)
for rows.Next() {
var metric domain.RPUsageDailyMetric
if err := rows.Scan(
&metric.Date,
&metric.TenantID,
&metric.TenantType,
&metric.ClientID,
&metric.ClientName,
&metric.LoginRequests,
&metric.OtherRequests,
&metric.UniqueSubjects,
); err != nil {
return nil, fmt.Errorf("failed to scan rp usage daily aggregate: %w", err)
}
if metric.ClientName == "" {
metric.ClientName = metric.ClientID
}
metrics = append(metrics, metric)
}
return metrics, nil
}
func (r *ClickHouseRepository) FindPage(ctx context.Context, limit int, cursor *domain.AuditCursor, tenantID string) ([]domain.AuditLog, error) {
if limit <= 0 {
limit = 50
@@ -228,6 +412,21 @@ func (r *ClickHouseRepository) CountFailuresSince(ctx context.Context, since tim
return count, nil
}
func (r *ClickHouseRepository) CountEventsSince(ctx context.Context, since time.Time) (int64, error) {
sinceUTC := since.UTC().Format("2006-01-02 15:04:05")
query := fmt.Sprintf(`
SELECT count()
FROM audit_logs
WHERE timestamp >= toDateTime('%s')
`, sinceUTC)
var count int64
err := r.conn.QueryRow(ctx, query).Scan(&count)
if err != nil {
return 0, fmt.Errorf("failed to count audit events: %w", err)
}
return count, nil
}
func (r *ClickHouseRepository) CountActiveSessionsSince(ctx context.Context, since time.Time, tenantID string) (int64, error) {
// We use uniqExact(session_id) to count unique sessions that had success events recently.
query := `

View File

@@ -63,7 +63,7 @@ func TestMain(m *testing.M) {
}
// Auto-migrate
err = db.AutoMigrate(&domain.Tenant{}, &domain.TenantDomain{}, &domain.User{}, &domain.ClientConsent{}, &domain.RPUserMetadata{})
err = db.AutoMigrate(&domain.Tenant{}, &domain.TenantDomain{}, &domain.User{}, &domain.ClientConsent{}, &domain.RPUserMetadata{}, &domain.RPUsageEvent{})
if err != nil {
log.Fatalf("failed to migrate database: %s", err)
}

View File

@@ -0,0 +1,91 @@
package repository
import (
"baron-sso-backend/internal/domain"
"context"
"time"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
type RPUsageOutboxRepository interface {
Create(ctx context.Context, event *domain.RPUsageEvent) error
ListReady(ctx context.Context, limit int) ([]domain.RPUsageEvent, error)
MarkProcessing(ctx context.Context, id string) error
MarkProcessed(ctx context.Context, id string) error
MarkFailed(ctx context.Context, id string, message string, nextAttemptAt time.Time) error
}
type rpUsageOutboxRepository struct {
db *gorm.DB
}
func NewRPUsageOutboxRepository(db *gorm.DB) RPUsageOutboxRepository {
return &rpUsageOutboxRepository{db: db}
}
func (r *rpUsageOutboxRepository) Create(ctx context.Context, event *domain.RPUsageEvent) error {
if event.Payload == nil {
event.Payload = domain.JSONMap{}
}
if event.Status == "" {
event.Status = domain.RPUsageOutboxStatusPending
}
if event.OccurredAt.IsZero() {
event.OccurredAt = time.Now()
}
return r.db.WithContext(ctx).Clauses(clause.OnConflict{
Columns: []clause.Column{{Name: "dedupe_key"}},
DoNothing: true,
}).Create(event).Error
}
func (r *rpUsageOutboxRepository) ListReady(ctx context.Context, limit int) ([]domain.RPUsageEvent, error) {
if limit <= 0 || limit > 100 {
limit = 20
}
var rows []domain.RPUsageEvent
err := r.db.WithContext(ctx).
Where("status = ? AND (next_attempt_at IS NULL OR next_attempt_at <= ?)", domain.RPUsageOutboxStatusPending, time.Now()).
Order("occurred_at asc, created_at asc").
Limit(limit).
Find(&rows).Error
return rows, err
}
func (r *rpUsageOutboxRepository) MarkProcessing(ctx context.Context, id string) error {
return r.db.WithContext(ctx).
Model(&domain.RPUsageEvent{}).
Where("id = ? AND status = ?", id, domain.RPUsageOutboxStatusPending).
Updates(map[string]any{
"status": domain.RPUsageOutboxStatusProcessing,
"updated_at": time.Now(),
}).Error
}
func (r *rpUsageOutboxRepository) MarkProcessed(ctx context.Context, id string) error {
now := time.Now()
return r.db.WithContext(ctx).
Model(&domain.RPUsageEvent{}).
Where("id = ?", id).
Updates(map[string]any{
"status": domain.RPUsageOutboxStatusProcessed,
"last_error": "",
"processed_at": &now,
"updated_at": now,
}).Error
}
func (r *rpUsageOutboxRepository) MarkFailed(ctx context.Context, id string, message string, nextAttemptAt time.Time) error {
return r.db.WithContext(ctx).
Model(&domain.RPUsageEvent{}).
Where("id = ?", id).
Updates(map[string]any{
"status": domain.RPUsageOutboxStatusFailed,
"retry_count": gorm.Expr("retry_count + 1"),
"last_error": message,
"next_attempt_at": &nextAttemptAt,
"updated_at": time.Now(),
}).Error
}

View File

@@ -0,0 +1,67 @@
package service
import (
"baron-sso-backend/internal/domain"
"baron-sso-backend/internal/repository"
"context"
"crypto/sha256"
"encoding/hex"
"fmt"
"strings"
"time"
)
type RPUsageEventEmitter struct {
repo repository.RPUsageOutboxRepository
}
func NewRPUsageEventEmitter(repo repository.RPUsageOutboxRepository) *RPUsageEventEmitter {
return &RPUsageEventEmitter{repo: repo}
}
func (e *RPUsageEventEmitter) EmitRPUsageEvent(ctx context.Context, event domain.RPUsageEvent) error {
if e == nil || e.repo == nil {
return nil
}
event.EventType = strings.TrimSpace(event.EventType)
event.Subject = strings.TrimSpace(event.Subject)
event.ClientID = strings.TrimSpace(event.ClientID)
event.Source = strings.TrimSpace(event.Source)
event.CorrelationID = strings.TrimSpace(event.CorrelationID)
if event.EventType == "" {
return fmt.Errorf("rp usage event type is required")
}
if event.Subject == "" {
return fmt.Errorf("rp usage subject is required")
}
if event.ClientID == "" {
return fmt.Errorf("rp usage client_id is required")
}
if event.Source == "" {
event.Source = "backend"
}
if event.OccurredAt.IsZero() {
event.OccurredAt = time.Now()
}
if event.DedupeKey == "" {
event.DedupeKey = buildRPUsageDedupeKey(event)
}
if event.Payload == nil {
event.Payload = domain.JSONMap{}
}
return e.repo.Create(ctx, &event)
}
func buildRPUsageDedupeKey(event domain.RPUsageEvent) string {
raw := strings.Join([]string{
event.EventType,
event.Subject,
event.ClientID,
event.SessionID,
event.Source,
event.CorrelationID,
event.OccurredAt.UTC().Format("2006-01-02T15:04:05.000Z"),
}, "|")
sum := sha256.Sum256([]byte(raw))
return hex.EncodeToString(sum[:])
}

View File

@@ -0,0 +1,132 @@
package service
import (
"baron-sso-backend/internal/domain"
"context"
"errors"
"testing"
"time"
"github.com/stretchr/testify/require"
)
type fakeRPUsageOutboxRepo struct {
created []domain.RPUsageEvent
ready []domain.RPUsageEvent
processing []string
processed []string
failed []string
createErr error
projectErr error
}
func (f *fakeRPUsageOutboxRepo) Create(ctx context.Context, event *domain.RPUsageEvent) error {
if f.createErr != nil {
return f.createErr
}
f.created = append(f.created, *event)
return nil
}
func (f *fakeRPUsageOutboxRepo) ListReady(ctx context.Context, limit int) ([]domain.RPUsageEvent, error) {
return f.ready, nil
}
func (f *fakeRPUsageOutboxRepo) MarkProcessing(ctx context.Context, id string) error {
f.processing = append(f.processing, id)
return nil
}
func (f *fakeRPUsageOutboxRepo) MarkProcessed(ctx context.Context, id string) error {
f.processed = append(f.processed, id)
return nil
}
func (f *fakeRPUsageOutboxRepo) MarkFailed(ctx context.Context, id string, message string, nextAttemptAt time.Time) error {
f.failed = append(f.failed, id)
return nil
}
type fakeRPUsageProjectionRepo struct {
created []domain.RPUsageEvent
err error
}
func (f *fakeRPUsageProjectionRepo) CreateRPUsageEvent(ctx context.Context, event domain.RPUsageEvent) error {
if f.err != nil {
return f.err
}
f.created = append(f.created, event)
return nil
}
func TestRPUsageEventEmitterRequiresCanonicalFields(t *testing.T) {
repo := &fakeRPUsageOutboxRepo{}
emitter := NewRPUsageEventEmitter(repo)
err := emitter.EmitRPUsageEvent(context.Background(), domain.RPUsageEvent{
EventType: domain.RPUsageEventTypeAuthorizationGranted,
ClientID: "client-app",
})
require.Error(t, err)
require.Empty(t, repo.created)
}
func TestRPUsageEventEmitterCreatesPendingOutboxEvent(t *testing.T) {
repo := &fakeRPUsageOutboxRepo{}
emitter := NewRPUsageEventEmitter(repo)
err := emitter.EmitRPUsageEvent(context.Background(), domain.RPUsageEvent{
EventType: domain.RPUsageEventTypeAuthorizationGranted,
Subject: "user-123",
ClientID: "client-app",
Source: "hydra_consent",
CorrelationID: "challenge-1",
})
require.NoError(t, err)
require.Len(t, repo.created, 1)
require.NotEmpty(t, repo.created[0].DedupeKey)
require.Equal(t, domain.RPUsageEventTypeAuthorizationGranted, repo.created[0].EventType)
require.Equal(t, "hydra_consent", repo.created[0].Source)
}
func TestRPUsageProjectorWorkerMarksProcessedAfterProjection(t *testing.T) {
outbox := &fakeRPUsageOutboxRepo{
ready: []domain.RPUsageEvent{{
ID: "event-1",
EventType: domain.RPUsageEventTypeAuthorizationGranted,
Subject: "user-123",
ClientID: "client-app",
}},
}
projection := &fakeRPUsageProjectionRepo{}
worker := NewRPUsageProjectorWorker(outbox, projection)
worker.processOnce(context.Background())
require.Equal(t, []string{"event-1"}, outbox.processing)
require.Equal(t, []string{"event-1"}, outbox.processed)
require.Empty(t, outbox.failed)
require.Len(t, projection.created, 1)
}
func TestRPUsageProjectorWorkerMarksFailedWhenProjectionFails(t *testing.T) {
outbox := &fakeRPUsageOutboxRepo{
ready: []domain.RPUsageEvent{{
ID: "event-1",
EventType: domain.RPUsageEventTypeAuthorizationGranted,
Subject: "user-123",
ClientID: "client-app",
}},
}
projection := &fakeRPUsageProjectionRepo{err: errors.New("clickhouse unavailable")}
worker := NewRPUsageProjectorWorker(outbox, projection)
worker.processOnce(context.Background())
require.Equal(t, []string{"event-1"}, outbox.processing)
require.Empty(t, outbox.processed)
require.Equal(t, []string{"event-1"}, outbox.failed)
}

View File

@@ -0,0 +1,82 @@
package service
import (
"baron-sso-backend/internal/domain"
"baron-sso-backend/internal/repository"
"context"
"log/slog"
"time"
)
type RPUsageProjectorWorker struct {
outbox repository.RPUsageOutboxRepository
projection domain.RPUsageProjectionRepository
interval time.Duration
batchSize int
}
func NewRPUsageProjectorWorker(outbox repository.RPUsageOutboxRepository, projection domain.RPUsageProjectionRepository) *RPUsageProjectorWorker {
return &RPUsageProjectorWorker{
outbox: outbox,
projection: projection,
interval: 5 * time.Second,
batchSize: 50,
}
}
func (w *RPUsageProjectorWorker) Start(ctx context.Context) {
if w == nil || w.outbox == nil || w.projection == nil {
return
}
ticker := time.NewTicker(w.interval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
return
default:
w.processOnce(ctx)
}
select {
case <-ctx.Done():
return
case <-ticker.C:
}
}
}
func (w *RPUsageProjectorWorker) processOnce(ctx context.Context) {
events, err := w.outbox.ListReady(ctx, w.batchSize)
if err != nil {
slog.Warn("failed to list rp usage outbox", "error", err)
return
}
for _, event := range events {
if err := w.outbox.MarkProcessing(ctx, event.ID); err != nil {
slog.Warn("failed to mark rp usage event processing", "event_id", event.ID, "error", err)
continue
}
if err := w.projection.CreateRPUsageEvent(ctx, event); err != nil {
nextAttempt := time.Now().Add(backoffDuration(event.RetryCount))
_ = w.outbox.MarkFailed(ctx, event.ID, err.Error(), nextAttempt)
slog.Warn("failed to project rp usage event", "event_id", event.ID, "error", err)
continue
}
if err := w.outbox.MarkProcessed(ctx, event.ID); err != nil {
slog.Warn("failed to mark rp usage event processed", "event_id", event.ID, "error", err)
}
}
}
func backoffDuration(retryCount int) time.Duration {
if retryCount < 0 {
retryCount = 0
}
delay := time.Duration(retryCount+1) * time.Minute
if delay > 30*time.Minute {
return 30 * time.Minute
}
return delay
}