perf: batch DB operations in ingest and plan list to cut CPU load

server/internal/api/ingest.go

@@ -3,7 +3,6 @@ package api
 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"net/http"
 	"strings"
 	"time"
@@ -104,33 +103,22 @@ func (h *IngestHandler) Handle(w http.ResponseWriter, r *http.Request) {
 		session.GitBranch = req.GitBranch
 	}
 
-	// Create events from transcript lines
-	eventsCreated := 0
+	// Build events from transcript lines. Title auto-generation happens first
+	// so the UPDATE runs once before the batch insert.
+	events := make([]*domain.Event, 0, len(req.TranscriptLines))
 	for _, line := range req.TranscriptLines {
-		event := &domain.Event{
-			SessionID: session.ID,
-			Payload:   line,
-		}
-
-		// Extract uuid from transcript line (Claude Code's unique identifier)
-		if uuid, ok := line["uuid"].(string); ok {
-			event.UUID = uuid
-		}
-
-		// Extract type if present
 		eventType := ""
 		if et, ok := line["type"].(string); ok {
 			eventType = et
-			event.EventType = et
 		}
 
 		// Skip events that should not be stored (high-volume, not needed for display)
 		if domain.IsSkippedEventType(eventType) {
 			continue
 		}
 
-		// Auto-generate title from first user message if not set
-		// Skip meta messages, command messages, and tool results
+		// Auto-generate title from first user message if not set.
+		// Skip meta messages, command messages, and tool results.
 		if eventType == "user" && session.Title == nil && !isMetaMessage(line) {
 			if text := extractUserMessageText(line); text != "" && isValidUserInput(text) {
 				title := truncateString(text, 45)
@@ -140,15 +128,23 @@ func (h *IngestHandler) Handle(w http.ResponseWriter, r *http.Request) {
 			}
 		}
 
-		if err := h.repos.Event.Create(ctx, event); err != nil {
-			// Skip duplicate events (same uuid within session)
-			if errors.Is(err, repository.ErrDuplicateEvent) {
-				continue
-			}
-			http.Error(w, `{"error": "failed to create event"}`, http.StatusInternalServerError)
-			return
+		event := &domain.Event{
+			SessionID: session.ID,
+			EventType: eventType,
+			Payload:   line,
 		}
-		eventsCreated++
+		if uuid, ok := line["uuid"].(string); ok {
+			event.UUID = uuid
+		}
+		events = append(events, event)
+	}
+
+	// Persist all events in a single batch. Duplicate UUIDs are silently skipped
+	// by the repository so retried hook invocations don't fail.
+	eventsCreated, err := h.repos.Event.CreateBatch(ctx, events)
+	if err != nil {
+		http.Error(w, `{"error": "failed to create event"}`, http.StatusInternalServerError)
+		return
 	}
 
 	// Update session's updated_at timestamp if events were created

server/internal/api/plan_document.go

@@ -106,44 +106,121 @@ func (h *PlanDocumentHandler) planDocumentToResponse(ctx context.Context, doc *d
 		return nil, err
 	}
 
-	// Fetch user details
-	collaborators := make([]*CollaboratorResponse, 0, len(userIDs))
-	for _, userID := range userIDs {
-		user, err := h.repos.User.FindByID(ctx, userID)
-		if err == nil && user != nil {
+	users, err := h.repos.User.FindByIDs(ctx, userIDs)
+	if err != nil {
+		return nil, err
+	}
+
+	// Fetch project (single lookup for single-plan paths)
+	var projectCache map[string]*domain.Project
+	if doc.ProjectID != "" {
+		projectCache, err = h.repos.Project.FindByIDs(ctx, []string{doc.ProjectID})
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	activeCommentsCount, err := h.repos.PlanCommentThread.CountActiveByPlanDocumentID(ctx, doc.ID)
+	if err != nil {
+		// Count is non-critical; fall back to zero rather than failing the request.
+		activeCommentsCount = 0
+	}
+
+	return h.buildPlanDocumentResponse(doc, isFavorited, userIDs, users, projectCache, activeCommentsCount), nil
+}
+
+// planDocumentsToResponseBatch builds responses for multiple plans while
+// pre-batching user and project lookups to eliminate N+1 queries during list
+// requests.
+func (h *PlanDocumentHandler) planDocumentsToResponseBatch(ctx context.Context, docs []*domain.PlanDocument, favoritedIDs map[string]bool) ([]*PlanDocumentResponse, error) {
+	if len(docs) == 0 {
+		return []*PlanDocumentResponse{}, nil
+	}
+
+	// Collect collaborator user IDs per plan (one query per plan; the expensive
+	// part — resolving each user — is batched below).
+	collaboratorIDsByPlan := make(map[string][]string, len(docs))
+	userIDSet := make(map[string]struct{})
+	projectIDSet := make(map[string]struct{})
+
+	for _, doc := range docs {
+		ids, err := h.repos.PlanDocumentEvent.GetCollaboratorUserIDs(ctx, doc.ID)
+		if err != nil {
+			return nil, err
+		}
+		collaboratorIDsByPlan[doc.ID] = ids
+		for _, id := range ids {
+			userIDSet[id] = struct{}{}
+		}
+		if doc.ProjectID != "" {
+			projectIDSet[doc.ProjectID] = struct{}{}
+		}
+	}
+
+	userIDs := make([]string, 0, len(userIDSet))
+	for id := range userIDSet {
+		userIDs = append(userIDs, id)
+	}
+	users, err := h.repos.User.FindByIDs(ctx, userIDs)
+	if err != nil {
+		return nil, err
+	}
+
+	projectIDs := make([]string, 0, len(projectIDSet))
+	for id := range projectIDSet {
+		projectIDs = append(projectIDs, id)
+	}
+	projects, err := h.repos.Project.FindByIDs(ctx, projectIDs)
+	if err != nil {
+		return nil, err
+	}
+
+	responses := make([]*PlanDocumentResponse, 0, len(docs))
+	for _, doc := range docs {
+		count, err := h.repos.PlanCommentThread.CountActiveByPlanDocumentID(ctx, doc.ID)
+		if err != nil {
+			count = 0
+		}
+		responses = append(responses, h.buildPlanDocumentResponse(
+			doc, favoritedIDs[doc.ID], collaboratorIDsByPlan[doc.ID], users, projects, count,
+		))
+	}
+	return responses, nil
+}
+
+func (h *PlanDocumentHandler) buildPlanDocumentResponse(
+	doc *domain.PlanDocument,
+	isFavorited bool,
+	collaboratorIDs []string,
+	users map[string]*domain.User,
+	projects map[string]*domain.Project,
+	activeCommentsCount int,
+) *PlanDocumentResponse {
+	collaborators := make([]*CollaboratorResponse, 0, len(collaboratorIDs))
+	for _, userID := range collaboratorIDs {
+		if user, ok := users[userID]; ok && user != nil {
 			collaborators = append(collaborators, &CollaboratorResponse{
 				ID:          user.ID,
 				DisplayName: user.GetDisplayName(),
 			})
 		}
 	}
 
-	// Get project info
 	var projectResp *PlanDocumentProjectResponse
 	if doc.ProjectID != "" {
-		project, err := h.repos.Project.FindByID(ctx, doc.ProjectID)
-		if err == nil && project != nil {
+		if project, ok := projects[doc.ProjectID]; ok && project != nil {
 			projectResp = &PlanDocumentProjectResponse{
 				ID:                     project.ID,
 				CanonicalGitRepository: project.CanonicalGitRepository,
 			}
 		}
 	}
 
-	// Generate URL if webURL is configured
 	var url string
 	if h.webURL != "" {
 		url = h.webURL + "/plans/" + doc.ID
 	}
 
-	// Count active comment threads
-	activeCommentsCount := 0
-	activeStatus := domain.PlanCommentThreadStatusActive
-	threads, err := h.repos.PlanCommentThread.FindByPlanDocumentID(ctx, doc.ID, &activeStatus)
-	if err == nil {
-		activeCommentsCount = len(threads)
-	}
-
 	return &PlanDocumentResponse{
 		ID:                  doc.ID,
 		Project:             projectResp,
@@ -156,7 +233,7 @@ func (h *PlanDocumentHandler) planDocumentToResponse(ctx context.Context, doc *d
 		UpdatedAt:           doc.UpdatedAt.Format("2006-01-02T15:04:05Z07:00"),
 		IsFavorited:         isFavorited,
 		ActiveCommentsCount: activeCommentsCount,
-	}, nil
+	}
 }
 
 func (h *PlanDocumentHandler) eventToResponse(ctx context.Context, event *domain.PlanDocumentEvent) *PlanDocumentEventResponse {
@@ -308,15 +385,10 @@ func (h *PlanDocumentHandler) List(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	plans := make([]*PlanDocumentResponse, 0, len(docs))
-	for _, doc := range docs {
-		isFavorited := favoritedIDs[doc.ID]
-		resp, err := h.planDocumentToResponse(ctx, doc, isFavorited)
-		if err != nil {
-			http.Error(w, `{"error": "failed to build response"}`, http.StatusInternalServerError)
-			return
-		}
-		plans = append(plans, resp)
+	plans, err := h.planDocumentsToResponseBatch(ctx, docs, favoritedIDs)
+	if err != nil {
+		http.Error(w, `{"error": "failed to build response"}`, http.StatusInternalServerError)
+		return
 	}
 
 	// Sort: favorited plans first, then by updated_at desc (already sorted by repo)

server/internal/repository/dynamodb/event.go

@@ -91,6 +91,26 @@ func (r *EventRepository) Create(ctx context.Context, event *domain.Event) error
 	return nil
 }
 
+// CreateBatch for DynamoDB falls back to serial Create calls because BatchWriteItem
+// does not support conditional expressions, which we need to skip duplicate UUIDs.
+// The API still exposes the batched interface so SQL backends can benefit from
+// transaction-level batching.
+func (r *EventRepository) CreateBatch(ctx context.Context, events []*domain.Event) (int, error) {
+	inserted := 0
+	for _, event := range events {
+		err := r.Create(ctx, event)
+		if err == nil {
+			inserted++
+			continue
+		}
+		if err == repository.ErrDuplicateEvent {
+			continue
+		}
+		return inserted, err
+	}
+	return inserted, nil
+}
+
 func (r *EventRepository) FindBySessionID(ctx context.Context, sessionID string) ([]*domain.Event, error) {
 	keyCond := expression.Key("session_id").Equal(expression.Value(sessionID))
 

server/internal/repository/dynamodb/plan_comment_thread.go

@@ -138,6 +138,29 @@ func (r *PlanCommentThreadRepository) FindByPlanDocumentID(ctx context.Context,
 	return threads, nil
 }
 
+func (r *PlanCommentThreadRepository) CountActiveByPlanDocumentID(ctx context.Context, planDocumentID string) (int, error) {
+	keyCond := expression.Key("plan_document_id").Equal(expression.Value(planDocumentID))
+	filter := expression.Name("status").Equal(expression.Value(string(domain.PlanCommentThreadStatusActive)))
+	expr, err := expression.NewBuilder().WithKeyCondition(keyCond).WithFilter(filter).Build()
+	if err != nil {
+		return 0, err
+	}
+
+	result, err := r.db.Client.Query(ctx, &dynamodb.QueryInput{
+		TableName:                 aws.String(r.db.TableName("plan_comment_threads")),
+		IndexName:                 aws.String("plan_document_id-created_at-index"),
+		KeyConditionExpression:    expr.KeyCondition(),
+		FilterExpression:          expr.Filter(),
+		ExpressionAttributeNames:  expr.Names(),
+		ExpressionAttributeValues: expr.Values(),
+		Select:                    types.SelectCount,
+	})
+	if err != nil {
+		return 0, err
+	}
+	return int(result.Count), nil
+}
+
 func (r *PlanCommentThreadRepository) Update(ctx context.Context, thread *domain.PlanCommentThread) error {
 	thread.UpdatedAt = time.Now()
 

server/internal/repository/dynamodb/project.go

@@ -78,6 +78,71 @@ func (r *ProjectRepository) FindByID(ctx context.Context, id string) (*domain.Pr
 	return r.itemToProject(&item), nil
 }
 
+func (r *ProjectRepository) FindByIDs(ctx context.Context, ids []string) (map[string]*domain.Project, error) {
+	result := make(map[string]*domain.Project, len(ids))
+	if len(ids) == 0 {
+		return result, nil
+	}
+
+	// Deduplicate to avoid duplicate keys in BatchGetItem (which would return an error).
+	seen := make(map[string]struct{}, len(ids))
+	unique := make([]string, 0, len(ids))
+	for _, id := range ids {
+		if _, ok := seen[id]; ok {
+			continue
+		}
+		seen[id] = struct{}{}
+		unique = append(unique, id)
+	}
+
+	tableName := r.db.TableName("projects")
+
+	// BatchGetItem is limited to 100 items per call.
+	const batchLimit = 100
+	for start := 0; start < len(unique); start += batchLimit {
+		end := start + batchLimit
+		if end > len(unique) {
+			end = len(unique)
+		}
+		keys := make([]map[string]types.AttributeValue, 0, end-start)
+		for _, id := range unique[start:end] {
+			keys = append(keys, map[string]types.AttributeValue{
+				"id": &types.AttributeValueMemberS{Value: id},
+			})
+		}
+
+		req := map[string]types.KeysAndAttributes{
+			tableName: {Keys: keys},
+		}
+
+		for len(req) > 0 {
+			output, err := r.db.Client.BatchGetItem(ctx, &dynamodb.BatchGetItemInput{
+				RequestItems: req,
+			})
+			if err != nil {
+				return nil, err
+			}
+
+			for _, raw := range output.Responses[tableName] {
+				var item projectItem
+				if err := attributevalue.UnmarshalMap(raw, &item); err != nil {
+					return nil, err
+				}
+				p := r.itemToProject(&item)
+				result[p.ID] = p
+			}
+
+			// DynamoDB may not process all items; retry the unprocessed ones.
+			if unprocessed, ok := output.UnprocessedKeys[tableName]; ok && len(unprocessed.Keys) > 0 {
+				req = map[string]types.KeysAndAttributes{tableName: unprocessed}
+			} else {
+				req = nil
+			}
+		}
+	}
+	return result, nil
+}
+
 func (r *ProjectRepository) FindByCanonicalGitRepository(ctx context.Context, canonicalGitRepo string) (*domain.Project, error) {
 	keyCond := expression.Key("canonical_git_repository").Equal(expression.Value(canonicalGitRepo))
 	expr, err := expression.NewBuilder().WithKeyCondition(keyCond).Build()