Merge remote-tracking branch 'origin/dev' into fix/issue-637

2026-05-07 13:53:14 +09:00
parent c398237c35 281b690c38
commit b540482bf5
103 changed files with 6332 additions and 1539 deletions
--- a/backend/internal/repository/clickhouse_repo.go
+++ b/backend/internal/repository/clickhouse_repo.go
@@ -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 := `
--- a/backend/internal/repository/main_test.go
+++ b/backend/internal/repository/main_test.go
@@ -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)
 	}
--- a/backend/internal/repository/rp_usage_outbox_repository.go
+++ b/backend/internal/repository/rp_usage_outbox_repository.go
@@ -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
+}