RFC-003: パフォーマンス最適化 - N+1問題の解決とキャッシュ戦略の改善¶

// BAD: N+1 query
func (s *EventService) GetEvents(calendarID string) ([]*Event, error) {
    events, err := s.db.Query("SELECT * FROM events WHERE calendar_id = $1", calendarID)
    if err != nil {
        return nil, err
    }

    for _, event := range events {
        // N+1: Each event triggers separate queries
        event.Attendees, _ = s.getAttendees(event.ID)      // Query 1, 2, 3, ...
        event.Resources, _ = s.getResources(event.ID)      // Query 1, 2, 3, ...
        event.Organizer, _ = s.getUserByID(event.OrganizerID)  // Query 1, 2, 3, ...
    }

    return events, nil
}

// Total queries: 1 + N*3 = 1 + 50*3 = 151 queries for 50 events!

// GOOD: Batched queries with DataLoader
type DataLoaders struct {
    AttendeeLoader *dataloader.Loader
    ResourceLoader *dataloader.Loader
    UserLoader     *dataloader.Loader
}

func (s *EventService) GetEvents(ctx context.Context, calendarID string) ([]*Event, error) {
    events, err := s.db.Query("SELECT * FROM events WHERE calendar_id = $1", calendarID)
    if err != nil {
        return nil, err
    }

    // Extract all IDs
    eventIDs := extractIDs(events)
    userIDs := extractUserIDs(events)

    // DataLoader batches and caches automatically
    loaders := GetDataLoaders(ctx)

    // Batch load attendees (1 query for all events)
    attendeesMap, _ := loaders.AttendeeLoader.LoadMany(ctx, eventIDs)

    // Batch load resources (1 query for all events)
    resourcesMap, _ := loaders.ResourceLoader.LoadMany(ctx, eventIDs)

    // Batch load users (1 query for all unique users)
    usersMap, _ := loaders.UserLoader.LoadMany(ctx, userIDs)

    // Assemble results
    for _, event := range events {
        event.Attendees = attendeesMap[event.ID]
        event.Resources = resourcesMap[event.ID]
        event.Organizer = usersMap[event.OrganizerID]
    }

    return events, nil
}

// Total queries: 1 + 3 = 4 queries for any number of events!
// 151 queries → 4 queries (97% reduction)

// AttendeeLoader
func NewAttendeeLoader(db *Database) *dataloader.Loader {
    return dataloader.NewBatchedLoader(
        func(ctx context.Context, eventIDs []string) []*dataloader.Result {
            // Single batch query
            query := `
                SELECT event_id, user_id, email, name, response_status, is_optional
                FROM attendees
                WHERE event_id = ANY($1)
                ORDER BY event_id, created_at
            `

            rows, err := db.Query(ctx, query, pq.Array(eventIDs))
            if err != nil {
                return errorResults(len(eventIDs), err)
            }

            // Group by event_id
            attendeesByEvent := make(map[string][]*Attendee)
            for rows.Next() {
                var a Attendee
                rows.Scan(&a.EventID, &a.UserID, &a.Email, &a.Name, &a.ResponseStatus, &a.IsOptional)
                attendeesByEvent[a.EventID] = append(attendeesByEvent[a.EventID], &a)
            }

            // Return results in the same order as keys
            results := make([]*dataloader.Result, len(eventIDs))
            for i, eventID := range eventIDs {
                results[i] = &dataloader.Result{Data: attendeesByEvent[eventID]}
            }
            return results
        },
        dataloader.WithCache(&dataloader.NoCache{}),  // Use Redis for caching
    )
}

// Poor connection pool settings
db, err := sql.Open("postgres", dsn)
db.SetMaxOpenConns(100)  // Too high, causes contention
db.SetMaxIdleConns(10)   // Too low, frequent reconnections
db.SetConnMaxLifetime(time.Hour)  // Too long

// Optimized connection pool
db, err := sql.Open("postgres", dsn)

// Formula: MaxOpenConns = ((core_count * 2) + effective_spindle_count)
// For our DB: 8 cores, SSD → (8*2 + 1) = 17
db.SetMaxOpenConns(20)  // Slightly above formula

// Keep idle connections warm (reduce latency)
db.SetMaxIdleConns(10)  // 50% of max

// Prevent stale connections
db.SetConnMaxLifetime(30 * time.Minute)

// Prevent connection leaks
db.SetConnMaxIdleTime(10 * time.Minute)

// Monitor pool stats
go monitorConnectionPool(db)

func monitorConnectionPool(db *sql.DB) {
    ticker := time.NewTicker(30 * time.Second)
    for range ticker.C {
        stats := db.Stats()

        metrics.Gauge("db.connections.open", stats.OpenConnections)
        metrics.Gauge("db.connections.in_use", stats.InUse)
        metrics.Gauge("db.connections.idle", stats.Idle)
        metrics.Counter("db.connections.wait_count", stats.WaitCount)
        metrics.Histogram("db.connections.wait_duration", stats.WaitDuration.Milliseconds())

        // Alert if pool is saturated
        if float64(stats.InUse)/float64(stats.MaxOpenConnections) > 0.9 {
            alerts.Warning("Database connection pool near saturation")
        }
    }
}

type CacheManager struct {
    l1 *sync.Map           // In-memory cache
    l2 *redis.Client       // Redis cache
    db *Database           // PostgreSQL
}

func (c *CacheManager) Get(ctx context.Context, key string) (interface{}, error) {
    // L1: Check in-memory cache
    if val, ok := c.l1.Load(key); ok {
        metrics.Counter("cache.l1.hit").Inc()
        return val, nil
    }
    metrics.Counter("cache.l1.miss").Inc()

    // L2: Check Redis
    val, err := c.l2.Get(ctx, key).Result()
    if err == nil {
        metrics.Counter("cache.l2.hit").Inc()

        // Populate L1
        c.l1.Store(key, val)
        time.AfterFunc(30*time.Second, func() { c.l1.Delete(key) })

        return val, nil
    }
    metrics.Counter("cache.l2.miss").Inc()

    // L3: Fetch from database
    data, err := c.db.Get(ctx, key)
    if err != nil {
        return nil, err
    }

    // Populate caches (write-through)
    c.Set(ctx, key, data, getTTL(key))

    return data, nil
}

func (c *CacheManager) Set(ctx context.Context, key string, value interface{}, ttl time.Duration) {
    // L1
    c.l1.Store(key, value)
    time.AfterFunc(30*time.Second, func() { c.l1.Delete(key) })

    // L2
    c.l2.Set(ctx, key, value, ttl)
}

func (c *CacheManager) GetWithPEE(ctx context.Context, key string, ttl time.Duration, fetchFn func() (interface{}, error)) (interface{}, error) {
    val, expiry, err := c.getWithExpiry(ctx, key)

    if err == nil {
        // Calculate probability of early refresh
        timeRemaining := expiry.Sub(time.Now())
        delta := ttl - timeRemaining
        beta := 1.0  // Tuning parameter

        // Probability increases as expiry approaches
        prob := -beta * math.Log(rand.Float64())

        if timeRemaining.Seconds() > prob*delta.Seconds() {
            // Use cached value
            return val, nil
        }
    }

    // Refresh cache (single-flight to prevent stampede)
    result, err, _ := c.singleflight.Do(key, func() (interface{}, error) {
        data, err := fetchFn()
        if err != nil {
            return nil, err
        }

        c.Set(ctx, key, data, ttl)
        return data, nil
    })

    return result, err
}

// When event is updated, invalidate related caches
func (s *EventService) UpdateEvent(ctx context.Context, eventID string, updates *EventUpdates) error {
    // Update database
    event, err := s.db.UpdateEvent(ctx, eventID, updates)
    if err != nil {
        return err
    }

    // Invalidate caches
    go s.invalidateCaches(ctx, event)

    // Publish invalidation event (for other service instances)
    s.kafka.Publish(ctx, "cache.invalidate", CacheInvalidationEvent{
        Keys: []string{
            fmt.Sprintf("event:%s:full", eventID),
            fmt.Sprintf("event:%s:attendees", eventID),
            fmt.Sprintf("calendar:%s:events:*", event.CalendarID),
        },
        Timestamp: time.Now(),
    })

    return nil
}

func (s *EventService) invalidateCaches(ctx context.Context, event *Event) {
    keys := []string{
        fmt.Sprintf("event:%s:full", event.ID),
        fmt.Sprintf("event:%s:attendees", event.ID),
    }

    // Invalidate all affected calendar caches
    pattern := fmt.Sprintf("calendar:%s:events:*", event.CalendarID)
    calendarKeys, _ := s.cache.Keys(ctx, pattern)
    keys = append(keys, calendarKeys...)

    // Delete in batch
    s.cache.Del(ctx, keys...)
}