From d9910ac5ae4db67e845cd7893d80bc94d3ce3c20 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 21 Apr 2026 09:16:55 +0000 Subject: [PATCH] perf: batch DB operations in ingest and plan list to cut CPU load The ingest endpoint previously issued one INSERT per transcript line, and the plan list endpoint issued a nested sequence of User.FindByID, Project.FindByID, and a full FindByPlanDocumentID scan per plan. Both compounded into O(N) round-trips on hot paths, pegging DB CPU. Repository additions: - EventRepository.CreateBatch: single transaction / multi-row INSERT - UserRepository.FindByIDs and ProjectRepository.FindByIDs - PlanCommentThreadRepository.CountActiveByPlanDocumentID (COUNT only) Handler changes: - /api/ingest now builds all events once and persists them in one call. Duplicate UUIDs are still silently skipped. - Plan list prefetches users and projects in two batch queries and counts comment threads via the new COUNT query instead of fetching full thread rows. Implemented for memory, sqlite, postgres, turso, and dynamodb backends, with matching testsuite coverage that runs across all implementations. --- server/internal/api/ingest.go | 46 +++---- server/internal/api/plan_document.go | 126 ++++++++++++++---- server/internal/repository/dynamodb/event.go | 20 +++ .../dynamodb/plan_comment_thread.go | 23 ++++ .../internal/repository/dynamodb/project.go | 65 +++++++++ server/internal/repository/dynamodb/user.go | 62 +++++++++ server/internal/repository/interface.go | 10 ++ server/internal/repository/memory/event.go | 27 ++++ .../repository/memory/plan_comment_thread.go | 13 ++ server/internal/repository/memory/project.go | 17 +++ server/internal/repository/memory/user.go | 17 +++ server/internal/repository/postgres/event.go | 61 +++++++++ .../postgres/plan_comment_thread.go | 12 ++ .../internal/repository/postgres/project.go | 33 +++++ server/internal/repository/postgres/user.go | 35 +++++ server/internal/repository/sqlite/event.go | 57 ++++++++ .../repository/sqlite/plan_comment_thread.go | 12 ++ server/internal/repository/sqlite/project.go | 32 +++++ server/internal/repository/sqlite/user.go | 36 +++++ .../repository/testsuite/event_suite.go | 61 +++++++++ .../testsuite/plan_comment_thread_suite.go | 42 ++++++ .../repository/testsuite/project_suite.go | 27 ++++ .../repository/testsuite/user_suite.go | 28 ++++ server/internal/repository/turso/event.go | 57 ++++++++ .../repository/turso/plan_comment_thread.go | 12 ++ server/internal/repository/turso/project.go | 32 +++++ server/internal/repository/turso/user.go | 36 +++++ 27 files changed, 947 insertions(+), 52 deletions(-) diff --git a/server/internal/api/ingest.go b/server/internal/api/ingest.go index 68405be..452fa57 100644 --- a/server/internal/api/ingest.go +++ b/server/internal/api/ingest.go @@ -3,7 +3,6 @@ package api import ( "context" "encoding/json" - "errors" "net/http" "strings" "time" @@ -104,24 +103,13 @@ 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) @@ -129,8 +117,8 @@ func (h *IngestHandler) Handle(w http.ResponseWriter, r *http.Request) { 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 diff --git a/server/internal/api/plan_document.go b/server/internal/api/plan_document.go index 4523d9c..148fb47 100644 --- a/server/internal/api/plan_document.go +++ b/server/internal/api/plan_document.go @@ -106,11 +106,99 @@ 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(), @@ -118,11 +206,9 @@ func (h *PlanDocumentHandler) planDocumentToResponse(ctx context.Context, doc *d } } - // 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, @@ -130,20 +216,11 @@ func (h *PlanDocumentHandler) planDocumentToResponse(ctx context.Context, doc *d } } - // 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) diff --git a/server/internal/repository/dynamodb/event.go b/server/internal/repository/dynamodb/event.go index c53e8ea..2e31fce 100644 --- a/server/internal/repository/dynamodb/event.go +++ b/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)) diff --git a/server/internal/repository/dynamodb/plan_comment_thread.go b/server/internal/repository/dynamodb/plan_comment_thread.go index 82839d7..d693079 100644 --- a/server/internal/repository/dynamodb/plan_comment_thread.go +++ b/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() diff --git a/server/internal/repository/dynamodb/project.go b/server/internal/repository/dynamodb/project.go index 513f3cd..07e7da8 100644 --- a/server/internal/repository/dynamodb/project.go +++ b/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() diff --git a/server/internal/repository/dynamodb/user.go b/server/internal/repository/dynamodb/user.go index 560fb5e..cec34be 100644 --- a/server/internal/repository/dynamodb/user.go +++ b/server/internal/repository/dynamodb/user.go @@ -77,6 +77,68 @@ func (r *UserRepository) FindByID(ctx context.Context, id string) (*domain.User, return r.itemToUser(&item), nil } +func (r *UserRepository) FindByIDs(ctx context.Context, ids []string) (map[string]*domain.User, error) { + result := make(map[string]*domain.User, len(ids)) + if len(ids) == 0 { + return result, nil + } + + 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("users") + + 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 userItem + if err := attributevalue.UnmarshalMap(raw, &item); err != nil { + return nil, err + } + u := r.itemToUser(&item) + result[u.ID] = u + } + + 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 *UserRepository) FindByEmail(ctx context.Context, email string) (*domain.User, error) { keyCond := expression.Key("email").Equal(expression.Value(email)) expr, err := expression.NewBuilder().WithKeyCondition(keyCond).Build() diff --git a/server/internal/repository/interface.go b/server/internal/repository/interface.go index 47316ea..69735a2 100644 --- a/server/internal/repository/interface.go +++ b/server/internal/repository/interface.go @@ -15,6 +15,7 @@ var ErrDuplicateEvent = errors.New("duplicate event: UUID already exists for thi type ProjectRepository interface { Create(ctx context.Context, project *domain.Project) error FindByID(ctx context.Context, id string) (*domain.Project, error) + FindByIDs(ctx context.Context, ids []string) (map[string]*domain.Project, error) FindByCanonicalGitRepository(ctx context.Context, canonicalGitRepo string) (*domain.Project, error) FindOrCreateByCanonicalGitRepository(ctx context.Context, canonicalGitRepo string) (*domain.Project, error) FindAll(ctx context.Context, limit int, cursor string) ([]*domain.Project, string, error) // Returns (projects, nextCursor, error) @@ -43,6 +44,11 @@ type SessionRepository interface { // EventRepository はイベントの永続化を担当する type EventRepository interface { Create(ctx context.Context, event *domain.Event) error + // CreateBatch inserts multiple events efficiently. Duplicate events (same UUID + // within a session) are silently skipped. Returns the number of newly inserted + // events. Implementations should use a single transaction or batch operation + // to avoid serial round-trips. + CreateBatch(ctx context.Context, events []*domain.Event) (int, error) FindBySessionID(ctx context.Context, sessionID string) ([]*domain.Event, error) CountBySessionID(ctx context.Context, sessionID string) (int, error) DeleteBySessionID(ctx context.Context, sessionID string) error @@ -52,6 +58,7 @@ type EventRepository interface { type UserRepository interface { Create(ctx context.Context, user *domain.User) error FindByID(ctx context.Context, id string) (*domain.User, error) + FindByIDs(ctx context.Context, ids []string) (map[string]*domain.User, error) FindByEmail(ctx context.Context, email string) (*domain.User, error) FindAll(ctx context.Context) ([]*domain.User, error) UpdateDisplayName(ctx context.Context, id string, displayName string) error @@ -126,6 +133,9 @@ type PlanCommentThreadRepository interface { Create(ctx context.Context, thread *domain.PlanCommentThread) error FindByID(ctx context.Context, id string) (*domain.PlanCommentThread, error) FindByPlanDocumentID(ctx context.Context, planDocumentID string, status *domain.PlanCommentThreadStatus) ([]*domain.PlanCommentThread, error) + // CountActiveByPlanDocumentID returns the number of active threads for a plan + // document without loading thread data. + CountActiveByPlanDocumentID(ctx context.Context, planDocumentID string) (int, error) Update(ctx context.Context, thread *domain.PlanCommentThread) error Delete(ctx context.Context, id string) error // MarkOutdatedByPlanDocumentID marks all active threads as outdated for the given plan document diff --git a/server/internal/repository/memory/event.go b/server/internal/repository/memory/event.go index 753d438..fa25e40 100644 --- a/server/internal/repository/memory/event.go +++ b/server/internal/repository/memory/event.go @@ -29,6 +29,10 @@ func (r *EventRepository) Create(ctx context.Context, event *domain.Event) error r.mu.Lock() defer r.mu.Unlock() + return r.createLocked(event) +} + +func (r *EventRepository) createLocked(event *domain.Event) error { if event.ID == "" { event.ID = uuid.New().String() } @@ -49,6 +53,29 @@ func (r *EventRepository) Create(ctx context.Context, event *domain.Event) error return nil } +func (r *EventRepository) CreateBatch(ctx context.Context, events []*domain.Event) (int, error) { + if len(events) == 0 { + return 0, nil + } + + r.mu.Lock() + defer r.mu.Unlock() + + inserted := 0 + for _, event := range events { + err := r.createLocked(event) + if err == nil { + inserted++ + continue + } + if err == repository.ErrDuplicateEvent { + continue + } + return inserted, err + } + return inserted, nil +} + // getTimestampFromPayload extracts timestamp from payload, falls back to CreatedAt func getTimestampFromPayload(e *domain.Event) time.Time { if ts, ok := e.Payload["timestamp"].(string); ok { diff --git a/server/internal/repository/memory/plan_comment_thread.go b/server/internal/repository/memory/plan_comment_thread.go index 9f0f5db..37ee748 100644 --- a/server/internal/repository/memory/plan_comment_thread.go +++ b/server/internal/repository/memory/plan_comment_thread.go @@ -77,6 +77,19 @@ func (r *PlanCommentThreadRepository) FindByPlanDocumentID(ctx context.Context, return threads, nil } +func (r *PlanCommentThreadRepository) CountActiveByPlanDocumentID(ctx context.Context, planDocumentID string) (int, error) { + r.mu.RLock() + defer r.mu.RUnlock() + + count := 0 + for _, t := range r.threads { + if t.PlanDocumentID == planDocumentID && t.Status == domain.PlanCommentThreadStatusActive { + count++ + } + } + return count, nil +} + func (r *PlanCommentThreadRepository) Update(ctx context.Context, thread *domain.PlanCommentThread) error { r.mu.Lock() defer r.mu.Unlock() diff --git a/server/internal/repository/memory/project.go b/server/internal/repository/memory/project.go index ee17434..222810c 100644 --- a/server/internal/repository/memory/project.go +++ b/server/internal/repository/memory/project.go @@ -58,6 +58,23 @@ func (r *ProjectRepository) FindByID(ctx context.Context, id string) (*domain.Pr return project, 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 + } + + r.mu.RLock() + defer r.mu.RUnlock() + + for _, id := range ids { + if p, ok := r.projects[id]; ok { + result[id] = p + } + } + return result, nil +} + func (r *ProjectRepository) FindByCanonicalGitRepository(ctx context.Context, canonicalGitRepo string) (*domain.Project, error) { r.mu.RLock() defer r.mu.RUnlock() diff --git a/server/internal/repository/memory/user.go b/server/internal/repository/memory/user.go index 1ddeacf..f678327 100644 --- a/server/internal/repository/memory/user.go +++ b/server/internal/repository/memory/user.go @@ -46,6 +46,23 @@ func (r *UserRepository) FindByID(ctx context.Context, id string) (*domain.User, return user, nil } +func (r *UserRepository) FindByIDs(ctx context.Context, ids []string) (map[string]*domain.User, error) { + result := make(map[string]*domain.User, len(ids)) + if len(ids) == 0 { + return result, nil + } + + r.mu.RLock() + defer r.mu.RUnlock() + + for _, id := range ids { + if user, ok := r.users[id]; ok { + result[id] = user + } + } + return result, nil +} + func (r *UserRepository) FindByEmail(ctx context.Context, email string) (*domain.User, error) { r.mu.RLock() defer r.mu.RUnlock() diff --git a/server/internal/repository/postgres/event.go b/server/internal/repository/postgres/event.go index 68385ef..a8c8907 100644 --- a/server/internal/repository/postgres/event.go +++ b/server/internal/repository/postgres/event.go @@ -5,6 +5,7 @@ import ( "database/sql" "encoding/json" "fmt" + "strconv" "strings" "time" @@ -56,6 +57,66 @@ func (r *EventRepository) Create(ctx context.Context, event *domain.Event) error return nil } +func (r *EventRepository) CreateBatch(ctx context.Context, events []*domain.Event) (int, error) { + if len(events) == 0 { + return 0, nil + } + + // Build multi-row INSERT with ON CONFLICT DO NOTHING so duplicate uuids are + // silently skipped. Use RETURNING id to count successful inserts. + valueGroups := make([]string, 0, len(events)) + args := make([]any, 0, len(events)*6) + argIdx := 1 + + for _, event := range events { + if event.ID == "" { + event.ID = uuid.New().String() + } + if event.CreatedAt.IsZero() { + event.CreatedAt = time.Now() + } + + payloadJSON, err := json.Marshal(event.Payload) + if err != nil { + return 0, err + } + + var uuidValue sql.NullString + if event.UUID != "" { + uuidValue = sql.NullString{String: event.UUID, Valid: true} + } + + placeholders := make([]string, 6) + for i := 0; i < 6; i++ { + placeholders[i] = "$" + strconv.Itoa(argIdx) + argIdx++ + } + valueGroups = append(valueGroups, "("+strings.Join(placeholders, ",")+")") + args = append(args, event.ID, event.SessionID, uuidValue, event.EventType, payloadJSON, event.CreatedAt) + } + + query := `INSERT INTO events (id, session_id, uuid, event_type, payload, created_at) + VALUES ` + strings.Join(valueGroups, ",") + ` + ON CONFLICT (session_id, uuid) DO NOTHING + RETURNING id` + + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return 0, err + } + defer rows.Close() + + inserted := 0 + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return inserted, err + } + inserted++ + } + return inserted, rows.Err() +} + func (r *EventRepository) FindBySessionID(ctx context.Context, sessionID string) ([]*domain.Event, error) { // Order by payload->>'timestamp' if available, otherwise by created_at query := fmt.Sprintf(`SELECT id, session_id, uuid, event_type, payload, created_at diff --git a/server/internal/repository/postgres/plan_comment_thread.go b/server/internal/repository/postgres/plan_comment_thread.go index 34e1985..708dc62 100644 --- a/server/internal/repository/postgres/plan_comment_thread.go +++ b/server/internal/repository/postgres/plan_comment_thread.go @@ -85,6 +85,18 @@ func (r *PlanCommentThreadRepository) FindByPlanDocumentID(ctx context.Context, return threads, nil } +func (r *PlanCommentThreadRepository) CountActiveByPlanDocumentID(ctx context.Context, planDocumentID string) (int, error) { + var count int + err := r.db.QueryRowContext(ctx, + `SELECT COUNT(*) FROM plan_comment_threads WHERE plan_document_id = $1 AND status = $2`, + planDocumentID, string(domain.PlanCommentThreadStatusActive), + ).Scan(&count) + if err != nil { + return 0, err + } + return count, nil +} + func (r *PlanCommentThreadRepository) Update(ctx context.Context, thread *domain.PlanCommentThread) error { thread.UpdatedAt = time.Now() diff --git a/server/internal/repository/postgres/project.go b/server/internal/repository/postgres/project.go index 33ae5e2..121008d 100644 --- a/server/internal/repository/postgres/project.go +++ b/server/internal/repository/postgres/project.go @@ -4,6 +4,8 @@ import ( "context" "database/sql" "fmt" + "strconv" + "strings" "time" "github.com/google/uuid" @@ -43,6 +45,37 @@ func (r *ProjectRepository) FindByID(ctx context.Context, id string) (*domain.Pr )) } +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 + } + + placeholders := make([]string, len(ids)) + args := make([]any, len(ids)) + for i, id := range ids { + placeholders[i] = "$" + strconv.Itoa(i+1) + args[i] = id + } + + query := `SELECT id, canonical_git_repository, created_at FROM projects WHERE id IN (` + + strings.Join(placeholders, ",") + `)` + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + for rows.Next() { + p, err := r.scanProjectFromRows(rows) + if err != nil { + return nil, err + } + result[p.ID] = p + } + return result, rows.Err() +} + func (r *ProjectRepository) FindByCanonicalGitRepository(ctx context.Context, canonicalGitRepo string) (*domain.Project, error) { return r.scanProject(r.db.QueryRowContext(ctx, `SELECT id, canonical_git_repository, created_at diff --git a/server/internal/repository/postgres/user.go b/server/internal/repository/postgres/user.go index e999b03..b3bd526 100644 --- a/server/internal/repository/postgres/user.go +++ b/server/internal/repository/postgres/user.go @@ -3,6 +3,8 @@ package postgres import ( "context" "database/sql" + "strconv" + "strings" "time" "github.com/google/uuid" @@ -52,6 +54,39 @@ func (r *UserRepository) FindByID(ctx context.Context, id string) (*domain.User, return &user, nil } +func (r *UserRepository) FindByIDs(ctx context.Context, ids []string) (map[string]*domain.User, error) { + result := make(map[string]*domain.User, len(ids)) + if len(ids) == 0 { + return result, nil + } + + placeholders := make([]string, len(ids)) + args := make([]any, len(ids)) + for i, id := range ids { + placeholders[i] = "$" + strconv.Itoa(i+1) + args[i] = id + } + + query := `SELECT id, email, display_name, created_at FROM users WHERE id IN (` + + strings.Join(placeholders, ",") + `)` + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + for rows.Next() { + var user domain.User + var displayName sql.NullString + if err := rows.Scan(&user.ID, &user.Email, &displayName, &user.CreatedAt); err != nil { + return nil, err + } + user.DisplayName = displayName.String + result[user.ID] = &user + } + return result, rows.Err() +} + func (r *UserRepository) FindByEmail(ctx context.Context, email string) (*domain.User, error) { var user domain.User var displayName sql.NullString diff --git a/server/internal/repository/sqlite/event.go b/server/internal/repository/sqlite/event.go index 6140fb3..fdb503f 100644 --- a/server/internal/repository/sqlite/event.go +++ b/server/internal/repository/sqlite/event.go @@ -55,6 +55,63 @@ func (r *EventRepository) Create(ctx context.Context, event *domain.Event) error return nil } +func (r *EventRepository) CreateBatch(ctx context.Context, events []*domain.Event) (int, error) { + if len(events) == 0 { + return 0, nil + } + + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return 0, err + } + defer tx.Rollback() + + stmt, err := tx.PrepareContext(ctx, + `INSERT INTO events (id, session_id, uuid, event_type, payload, created_at) + VALUES (?, ?, ?, ?, ?, ?)`) + if err != nil { + return 0, err + } + defer stmt.Close() + + inserted := 0 + for _, event := range events { + if event.ID == "" { + event.ID = uuid.New().String() + } + if event.CreatedAt.IsZero() { + event.CreatedAt = time.Now() + } + + payloadJSON, err := json.Marshal(event.Payload) + if err != nil { + return inserted, err + } + + var uuidValue sql.NullString + if event.UUID != "" { + uuidValue = sql.NullString{String: event.UUID, Valid: true} + } + + _, err = stmt.ExecContext(ctx, + event.ID, event.SessionID, uuidValue, event.EventType, + string(payloadJSON), event.CreatedAt.Format(time.RFC3339Nano), + ) + if err != nil { + if strings.Contains(err.Error(), "UNIQUE constraint failed") { + continue + } + return inserted, err + } + inserted++ + } + + if err := tx.Commit(); err != nil { + return 0, err + } + return inserted, nil +} + func (r *EventRepository) FindBySessionID(ctx context.Context, sessionID string) ([]*domain.Event, error) { query := fmt.Sprintf(`SELECT id, session_id, uuid, event_type, payload, created_at FROM events WHERE session_id = ? diff --git a/server/internal/repository/sqlite/plan_comment_thread.go b/server/internal/repository/sqlite/plan_comment_thread.go index b710c8e..65ac813 100644 --- a/server/internal/repository/sqlite/plan_comment_thread.go +++ b/server/internal/repository/sqlite/plan_comment_thread.go @@ -85,6 +85,18 @@ func (r *PlanCommentThreadRepository) FindByPlanDocumentID(ctx context.Context, return threads, nil } +func (r *PlanCommentThreadRepository) CountActiveByPlanDocumentID(ctx context.Context, planDocumentID string) (int, error) { + var count int + err := r.db.QueryRowContext(ctx, + `SELECT COUNT(*) FROM plan_comment_threads WHERE plan_document_id = ? AND status = ?`, + planDocumentID, string(domain.PlanCommentThreadStatusActive), + ).Scan(&count) + if err != nil { + return 0, err + } + return count, nil +} + func (r *PlanCommentThreadRepository) Update(ctx context.Context, thread *domain.PlanCommentThread) error { thread.UpdatedAt = time.Now() diff --git a/server/internal/repository/sqlite/project.go b/server/internal/repository/sqlite/project.go index df68552..f462107 100644 --- a/server/internal/repository/sqlite/project.go +++ b/server/internal/repository/sqlite/project.go @@ -3,6 +3,7 @@ package sqlite import ( "context" "database/sql" + "strings" "time" "github.com/google/uuid" @@ -42,6 +43,37 @@ func (r *ProjectRepository) FindByID(ctx context.Context, id string) (*domain.Pr )) } +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 + } + + placeholders := make([]string, len(ids)) + args := make([]any, len(ids)) + for i, id := range ids { + placeholders[i] = "?" + args[i] = id + } + + query := `SELECT id, canonical_git_repository, created_at FROM projects WHERE id IN (` + + strings.Join(placeholders, ",") + `)` + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + for rows.Next() { + p, err := r.scanProjectFromRows(rows) + if err != nil { + return nil, err + } + result[p.ID] = p + } + return result, rows.Err() +} + func (r *ProjectRepository) FindByCanonicalGitRepository(ctx context.Context, canonicalGitRepo string) (*domain.Project, error) { return r.scanProject(r.db.QueryRowContext(ctx, `SELECT id, canonical_git_repository, created_at diff --git a/server/internal/repository/sqlite/user.go b/server/internal/repository/sqlite/user.go index 83b6885..e171524 100644 --- a/server/internal/repository/sqlite/user.go +++ b/server/internal/repository/sqlite/user.go @@ -3,6 +3,7 @@ package sqlite import ( "context" "database/sql" + "strings" "time" "github.com/google/uuid" @@ -54,6 +55,41 @@ func (r *UserRepository) FindByID(ctx context.Context, id string) (*domain.User, return &user, nil } +func (r *UserRepository) FindByIDs(ctx context.Context, ids []string) (map[string]*domain.User, error) { + result := make(map[string]*domain.User, len(ids)) + if len(ids) == 0 { + return result, nil + } + + placeholders := make([]string, len(ids)) + args := make([]any, len(ids)) + for i, id := range ids { + placeholders[i] = "?" + args[i] = id + } + + query := `SELECT id, email, display_name, created_at FROM users WHERE id IN (` + + strings.Join(placeholders, ",") + `)` + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + for rows.Next() { + var user domain.User + var createdAt string + var displayName sql.NullString + if err := rows.Scan(&user.ID, &user.Email, &displayName, &createdAt); err != nil { + return nil, err + } + user.DisplayName = displayName.String + user.CreatedAt, _ = time.Parse(time.RFC3339Nano, createdAt) + result[user.ID] = &user + } + return result, rows.Err() +} + func (r *UserRepository) FindByEmail(ctx context.Context, email string) (*domain.User, error) { var user domain.User var createdAt string diff --git a/server/internal/repository/testsuite/event_suite.go b/server/internal/repository/testsuite/event_suite.go index 6a4851f..696a2a6 100644 --- a/server/internal/repository/testsuite/event_suite.go +++ b/server/internal/repository/testsuite/event_suite.go @@ -465,6 +465,67 @@ func (s *EventRepositorySuite) TestCreate_DuplicateUUID_Concurrent() { s.Empty(otherErrors, "no unexpected errors should occur") } +func (s *EventRepositorySuite) TestCreateBatch() { + ctx := context.Background() + + sessionID := s.createTestSession("event-batch") + if sessionID == "" { + s.T().Skip("SessionRepo not available, skipping test") + } + + events := []*domain.Event{ + {SessionID: sessionID, UUID: uuid.New().String(), EventType: "user", Payload: map[string]interface{}{"n": 1}}, + {SessionID: sessionID, UUID: uuid.New().String(), EventType: "assistant", Payload: map[string]interface{}{"n": 2}}, + {SessionID: sessionID, UUID: uuid.New().String(), EventType: "user", Payload: map[string]interface{}{"n": 3}}, + } + + inserted, err := s.Repo.CreateBatch(ctx, events) + s.Require().NoError(err) + s.Equal(3, inserted) + + count, err := s.Repo.CountBySessionID(ctx, sessionID) + s.Require().NoError(err) + s.Equal(3, count) +} + +func (s *EventRepositorySuite) TestCreateBatch_SkipsDuplicates() { + ctx := context.Background() + + sessionID := s.createTestSession("event-batch-dup") + if sessionID == "" { + s.T().Skip("SessionRepo not available, skipping test") + } + + existing := &domain.Event{ + SessionID: sessionID, + UUID: uuid.New().String(), + EventType: "user", + Payload: map[string]interface{}{"first": true}, + } + s.Require().NoError(s.Repo.Create(ctx, existing)) + + events := []*domain.Event{ + {SessionID: sessionID, UUID: existing.UUID, EventType: "user", Payload: map[string]interface{}{"dup": true}}, + {SessionID: sessionID, UUID: uuid.New().String(), EventType: "assistant", Payload: map[string]interface{}{"n": 2}}, + } + + inserted, err := s.Repo.CreateBatch(ctx, events) + s.Require().NoError(err) + s.Equal(1, inserted, "only the non-duplicate event should be inserted") + + count, err := s.Repo.CountBySessionID(ctx, sessionID) + s.Require().NoError(err) + s.Equal(2, count) +} + +func (s *EventRepositorySuite) TestCreateBatch_Empty() { + ctx := context.Background() + + inserted, err := s.Repo.CreateBatch(ctx, nil) + s.Require().NoError(err) + s.Equal(0, inserted) +} + // isSQLiteBusyError checks if the error is a SQLite SQLITE_BUSY error func isSQLiteBusyError(err error) bool { if err == nil { diff --git a/server/internal/repository/testsuite/plan_comment_thread_suite.go b/server/internal/repository/testsuite/plan_comment_thread_suite.go index d81bf61..70decab 100644 --- a/server/internal/repository/testsuite/plan_comment_thread_suite.go +++ b/server/internal/repository/testsuite/plan_comment_thread_suite.go @@ -215,6 +215,48 @@ func (s *PlanCommentThreadRepositorySuite) TestFindByPlanDocumentID_WithStatusFi s.Len(outdatedThreads, 1) } +func (s *PlanCommentThreadRepositorySuite) TestCountActiveByPlanDocumentID() { + ctx := context.Background() + + planDocID := s.createTestPlanDocument("thread-count-active") + if planDocID == "" { + s.T().Skip("PlanDocRepo not available, skipping test") + } + + statuses := []domain.PlanCommentThreadStatus{ + domain.PlanCommentThreadStatusActive, + domain.PlanCommentThreadStatusActive, + domain.PlanCommentThreadStatusActive, + domain.PlanCommentThreadStatusResolved, + domain.PlanCommentThreadStatusOutdated, + } + for i, status := range statuses { + thread := &domain.PlanCommentThread{ + PlanDocumentID: planDocID, + TargetText: "count target " + string(rune('a'+i)), + Status: status, + } + s.Require().NoError(s.Repo.Create(ctx, thread)) + } + + count, err := s.Repo.CountActiveByPlanDocumentID(ctx, planDocID) + s.Require().NoError(err) + s.Equal(3, count) +} + +func (s *PlanCommentThreadRepositorySuite) TestCountActiveByPlanDocumentID_Empty() { + ctx := context.Background() + + planDocID := s.createTestPlanDocument("thread-count-empty") + if planDocID == "" { + s.T().Skip("PlanDocRepo not available, skipping test") + } + + count, err := s.Repo.CountActiveByPlanDocumentID(ctx, planDocID) + s.Require().NoError(err) + s.Equal(0, count) +} + func (s *PlanCommentThreadRepositorySuite) TestUpdate() { ctx := context.Background() diff --git a/server/internal/repository/testsuite/project_suite.go b/server/internal/repository/testsuite/project_suite.go index f3de5ba..2278461 100644 --- a/server/internal/repository/testsuite/project_suite.go +++ b/server/internal/repository/testsuite/project_suite.go @@ -100,6 +100,33 @@ func (s *ProjectRepositorySuite) TestFindByID_NotFound() { s.Nil(found) } +func (s *ProjectRepositorySuite) TestFindByIDs() { + ctx := context.Background() + + p1 := &domain.Project{CanonicalGitRepository: "https://github.com/example/batch1"} + p2 := &domain.Project{CanonicalGitRepository: "https://github.com/example/batch2"} + p3 := &domain.Project{CanonicalGitRepository: "https://github.com/example/batch3"} + s.Require().NoError(s.Repo.Create(ctx, p1)) + s.Require().NoError(s.Repo.Create(ctx, p2)) + s.Require().NoError(s.Repo.Create(ctx, p3)) + + result, err := s.Repo.FindByIDs(ctx, []string{p1.ID, p3.ID, p1.ID}) + s.Require().NoError(err) + s.Len(result, 2) + s.Equal(p1.CanonicalGitRepository, result[p1.ID].CanonicalGitRepository) + s.Equal(p3.CanonicalGitRepository, result[p3.ID].CanonicalGitRepository) + _, hasP2 := result[p2.ID] + s.False(hasP2) +} + +func (s *ProjectRepositorySuite) TestFindByIDs_Empty() { + ctx := context.Background() + + result, err := s.Repo.FindByIDs(ctx, nil) + s.Require().NoError(err) + s.Empty(result) +} + func (s *ProjectRepositorySuite) TestFindByCanonicalGitRepository() { ctx := context.Background() diff --git a/server/internal/repository/testsuite/user_suite.go b/server/internal/repository/testsuite/user_suite.go index 7ad8997..92e78ce 100644 --- a/server/internal/repository/testsuite/user_suite.go +++ b/server/internal/repository/testsuite/user_suite.go @@ -100,6 +100,34 @@ func (s *UserRepositorySuite) TestFindByEmail() { s.Equal(user.ID, found.ID) } +func (s *UserRepositorySuite) TestFindByIDs() { + ctx := context.Background() + + u1 := &domain.User{Email: "batch1@example.com", DisplayName: "One"} + u2 := &domain.User{Email: "batch2@example.com", DisplayName: "Two"} + u3 := &domain.User{Email: "batch3@example.com", DisplayName: "Three"} + s.Require().NoError(s.Repo.Create(ctx, u1)) + s.Require().NoError(s.Repo.Create(ctx, u2)) + s.Require().NoError(s.Repo.Create(ctx, u3)) + + // Pass a duplicate id to ensure the implementation handles it. + result, err := s.Repo.FindByIDs(ctx, []string{u1.ID, u3.ID, u1.ID}) + s.Require().NoError(err) + s.Len(result, 2) + s.Equal(u1.Email, result[u1.ID].Email) + s.Equal(u3.Email, result[u3.ID].Email) + _, hasU2 := result[u2.ID] + s.False(hasU2) +} + +func (s *UserRepositorySuite) TestFindByIDs_Empty() { + ctx := context.Background() + + result, err := s.Repo.FindByIDs(ctx, nil) + s.Require().NoError(err) + s.Empty(result) +} + func (s *UserRepositorySuite) TestFindByEmail_NotFound() { ctx := context.Background() diff --git a/server/internal/repository/turso/event.go b/server/internal/repository/turso/event.go index 05409cc..b5ba029 100644 --- a/server/internal/repository/turso/event.go +++ b/server/internal/repository/turso/event.go @@ -55,6 +55,63 @@ func (r *EventRepository) Create(ctx context.Context, event *domain.Event) error return nil } +func (r *EventRepository) CreateBatch(ctx context.Context, events []*domain.Event) (int, error) { + if len(events) == 0 { + return 0, nil + } + + tx, err := r.db.BeginTx(ctx, nil) + if err != nil { + return 0, err + } + defer tx.Rollback() + + stmt, err := tx.PrepareContext(ctx, + `INSERT INTO events (id, session_id, uuid, event_type, payload, created_at) + VALUES (?, ?, ?, ?, ?, ?)`) + if err != nil { + return 0, err + } + defer stmt.Close() + + inserted := 0 + for _, event := range events { + if event.ID == "" { + event.ID = uuid.New().String() + } + if event.CreatedAt.IsZero() { + event.CreatedAt = time.Now() + } + + payloadJSON, err := json.Marshal(event.Payload) + if err != nil { + return inserted, err + } + + var uuidValue sql.NullString + if event.UUID != "" { + uuidValue = sql.NullString{String: event.UUID, Valid: true} + } + + _, err = stmt.ExecContext(ctx, + event.ID, event.SessionID, uuidValue, event.EventType, + string(payloadJSON), event.CreatedAt.Format(time.RFC3339Nano), + ) + if err != nil { + if strings.Contains(err.Error(), "UNIQUE constraint failed") { + continue + } + return inserted, err + } + inserted++ + } + + if err := tx.Commit(); err != nil { + return 0, err + } + return inserted, nil +} + func (r *EventRepository) FindBySessionID(ctx context.Context, sessionID string) ([]*domain.Event, error) { query := fmt.Sprintf(`SELECT id, session_id, uuid, event_type, payload, created_at FROM events WHERE session_id = ? diff --git a/server/internal/repository/turso/plan_comment_thread.go b/server/internal/repository/turso/plan_comment_thread.go index 1ae52a4..fdc51d4 100644 --- a/server/internal/repository/turso/plan_comment_thread.go +++ b/server/internal/repository/turso/plan_comment_thread.go @@ -85,6 +85,18 @@ func (r *PlanCommentThreadRepository) FindByPlanDocumentID(ctx context.Context, return threads, nil } +func (r *PlanCommentThreadRepository) CountActiveByPlanDocumentID(ctx context.Context, planDocumentID string) (int, error) { + var count int + err := r.db.QueryRowContext(ctx, + `SELECT COUNT(*) FROM plan_comment_threads WHERE plan_document_id = ? AND status = ?`, + planDocumentID, string(domain.PlanCommentThreadStatusActive), + ).Scan(&count) + if err != nil { + return 0, err + } + return count, nil +} + func (r *PlanCommentThreadRepository) Update(ctx context.Context, thread *domain.PlanCommentThread) error { thread.UpdatedAt = time.Now() diff --git a/server/internal/repository/turso/project.go b/server/internal/repository/turso/project.go index cc9619c..9e6858f 100644 --- a/server/internal/repository/turso/project.go +++ b/server/internal/repository/turso/project.go @@ -3,6 +3,7 @@ package turso import ( "context" "database/sql" + "strings" "time" "github.com/google/uuid" @@ -42,6 +43,37 @@ func (r *ProjectRepository) FindByID(ctx context.Context, id string) (*domain.Pr )) } +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 + } + + placeholders := make([]string, len(ids)) + args := make([]any, len(ids)) + for i, id := range ids { + placeholders[i] = "?" + args[i] = id + } + + query := `SELECT id, canonical_git_repository, created_at FROM projects WHERE id IN (` + + strings.Join(placeholders, ",") + `)` + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + for rows.Next() { + p, err := r.scanProjectFromRows(rows) + if err != nil { + return nil, err + } + result[p.ID] = p + } + return result, rows.Err() +} + func (r *ProjectRepository) FindByCanonicalGitRepository(ctx context.Context, canonicalGitRepo string) (*domain.Project, error) { return r.scanProject(r.db.QueryRowContext(ctx, `SELECT id, canonical_git_repository, created_at diff --git a/server/internal/repository/turso/user.go b/server/internal/repository/turso/user.go index b64bb98..154a790 100644 --- a/server/internal/repository/turso/user.go +++ b/server/internal/repository/turso/user.go @@ -3,6 +3,7 @@ package turso import ( "context" "database/sql" + "strings" "time" "github.com/google/uuid" @@ -54,6 +55,41 @@ func (r *UserRepository) FindByID(ctx context.Context, id string) (*domain.User, return &user, nil } +func (r *UserRepository) FindByIDs(ctx context.Context, ids []string) (map[string]*domain.User, error) { + result := make(map[string]*domain.User, len(ids)) + if len(ids) == 0 { + return result, nil + } + + placeholders := make([]string, len(ids)) + args := make([]any, len(ids)) + for i, id := range ids { + placeholders[i] = "?" + args[i] = id + } + + query := `SELECT id, email, display_name, created_at FROM users WHERE id IN (` + + strings.Join(placeholders, ",") + `)` + rows, err := r.db.QueryContext(ctx, query, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + for rows.Next() { + var user domain.User + var createdAt string + var displayName sql.NullString + if err := rows.Scan(&user.ID, &user.Email, &displayName, &createdAt); err != nil { + return nil, err + } + user.DisplayName = displayName.String + user.CreatedAt, _ = time.Parse(time.RFC3339Nano, createdAt) + result[user.ID] = &user + } + return result, rows.Err() +} + func (r *UserRepository) FindByEmail(ctx context.Context, email string) (*domain.User, error) { var user domain.User var createdAt string