feat: adjust event processing to batch mode

- merge the processing logic for adding and modifying events
- merge the processing logic for renaming and deleting events
- optimize the batch event processing flow
- refactor the event processing failure handling logic

Author: WayneWang00

cmd/main.go

@@ -136,7 +136,7 @@ func main() {
 	schedulerService := service.NewScheduler(syncRepo, scanRepo, storageManager, appLogger)
 	fileScanService := service.NewFileScanService(workspaceRepo, eventRepo, scanRepo, storageManager, codebaseEmbeddingRepo, appLogger)
 	uploadService := service.NewUploadService(schedulerService, syncRepo, appLogger, syncServiceConfig)
-	embeddingProcessService := service.NewEmbeddingProcessService(workspaceRepo, eventRepo, codebaseEmbeddingRepo, uploadService, appLogger)
+	embeddingProcessService := service.NewEmbeddingProcessService(workspaceRepo, eventRepo, codebaseEmbeddingRepo, uploadService, syncRepo, appLogger)
 	embeddingStatusService := service.NewEmbeddingStatusService(codebaseEmbeddingRepo, workspaceRepo, eventRepo, syncRepo, appLogger)
 
 	// 创建存储
@@ -168,7 +168,7 @@ func main() {
 
 	// Initialize job layer
 	fileScanJob := job.NewFileScanJob(fileScanService, storageManager, syncRepo, appLogger, 5*time.Minute)
-	eventProcessorJob := job.NewEventProcessorJob(appLogger, syncRepo, embeddingProcessService, codegraphProcessor, storageManager)
+	eventProcessorJob := job.NewEventProcessorJob(appLogger, syncRepo, embeddingProcessService, codegraphProcessor, 10*time.Second, storageManager)
 	statusCheckerJob := job.NewStatusCheckerJob(embeddingStatusService, storageManager, syncRepo, appLogger, 5*time.Second)
 	eventCleanerJob := job.NewEventCleanerJob(eventRepo, appLogger)
 	indexCleanJob := job.NewIndexCleanJob(appLogger, indexer, workspaceRepo)

internal/job/event_processor_job.go

@@ -13,13 +13,14 @@ import (
 
 // EventProcessorJob 事件处理任务
 type EventProcessorJob struct {
-	httpSync  repository.SyncInterface
-	embedding service.EmbeddingProcessService
-	codegraph service.CodegraphProcessService
-	storage   repository.StorageInterface
-	logger    logger.Logger
-	ctx       context.Context
-	cancel    context.CancelFunc
+	httpSync          repository.SyncInterface
+	embedding         service.EmbeddingProcessService
+	codegraph         service.CodegraphProcessService
+	storage           repository.StorageInterface
+	logger            logger.Logger
+	embeddingInterval time.Duration
+	ctx               context.Context
+	cancel            context.CancelFunc
 }
 
 // NewEventProcessorJob 创建事件处理任务
@@ -28,17 +29,19 @@ func NewEventProcessorJob(
 	httpSync repository.SyncInterface,
 	embedding service.EmbeddingProcessService,
 	codegraph service.CodegraphProcessService,
+	embeddingInterval time.Duration,
 	storage repository.StorageInterface,
 ) *EventProcessorJob {
 	ctx, cancel := context.WithCancel(context.Background())
 	return &EventProcessorJob{
-		httpSync:  httpSync,
-		embedding: embedding,
-		codegraph: codegraph,
-		storage:   storage,
-		logger:    logger,
-		ctx:       ctx,
-		cancel:    cancel,
+		httpSync:          httpSync,
+		embedding:         embedding,
+		codegraph:         codegraph,
+		storage:           storage,
+		logger:            logger,
+		embeddingInterval: embeddingInterval,
+		ctx:               ctx,
+		cancel:            cancel,
 	}
 }
 
@@ -53,7 +56,7 @@ func (j *EventProcessorJob) Start(ctx context.Context) {
 
 	// 立即执行一次事件处理
 	authInfo := config.GetAuthInfo()
-	if authInfo.ClientId != "" && authInfo.Token != "" && authInfo.ServerURL != "" {
+	if j.embeddingInterval > 0 && authInfo.ClientId != "" && authInfo.Token != "" && authInfo.ServerURL != "" {
 		j.embeddingProcessWorkspaces(ctx)
 	}
 
@@ -63,7 +66,7 @@ func (j *EventProcessorJob) Start(ctx context.Context) {
 				j.logger.Error("recovered from panic in embedding processor: %v", r)
 			}
 		}()
-		ticker := time.NewTicker(5 * time.Second)
+		ticker := time.NewTicker(j.embeddingInterval)
 		defer ticker.Stop()
 
 		for {

internal/repository/event.go

@@ -49,6 +49,10 @@ type EventRepository interface {
 	BatchCreateEvents(events []*model.Event) error
 	// BatchDeleteEvents 批量删除事件
 	BatchDeleteEvents(ids []int64) error
+	// UpdateEvents 批量更新事件嵌入信息
+	UpdateEventsEmbedding(events []*model.Event) error
+	// UpdateEventsEmbeddingStatus 批量更新事件嵌入状态
+	UpdateEventsEmbeddingStatus(eventIDs []int64, status int) error
 	// GetExpiredEventIDs 获取过期事件的ID列表
 	GetExpiredEventIDs(cutoffTime time.Time) ([]int64, error)
 	// GetTableName 获取表名
@@ -1359,6 +1363,102 @@ func (r *eventRepository) BatchDeleteEvents(ids []int64) error {
 	return nil
 }
 
+// UpdateEvents 批量更新事件嵌入信息
+func (r *eventRepository) UpdateEventsEmbedding(events []*model.Event) error {
+	if len(events) == 0 {
+		return nil
+	}
+
+	tx, err := r.db.GetDB().Begin()
+	if err != nil {
+		return fmt.Errorf("failed to begin transaction: %w", err)
+	}
+	defer func() {
+		if err != nil {
+			tx.Rollback()
+		}
+	}()
+
+	query := `
+		UPDATE events
+		SET embedding_status = ?, sync_id = ?, file_hash = ?, updated_at = ?
+		WHERE id = ?
+	`
+
+	nowTime := time.Now()
+	stmt, err := tx.Prepare(query)
+	if err != nil {
+		return fmt.Errorf("failed to prepare statement: %w", err)
+	}
+	defer stmt.Close()
+
+	for _, event := range events {
+		_, err = stmt.Exec(
+			event.EmbeddingStatus,
+			event.SyncId,
+			event.FileHash,
+			nowTime,
+			event.ID,
+		)
+		if err != nil {
+			r.logger.Error("Failed to update event %d: %v", event.ID, err)
+			return fmt.Errorf("failed to update event %d: %w", event.ID, err)
+		}
+	}
+
+	if err = tx.Commit(); err != nil {
+		return fmt.Errorf("failed to commit transaction: %w", err)
+	}
+
+	r.logger.Info("Successfully updated %d events", len(events))
+	return nil
+}
+
+// UpdateEventsEmbeddingStatus 批量更新事件嵌入状态
+func (r *eventRepository) UpdateEventsEmbeddingStatus(eventIDs []int64, status int) error {
+	if len(eventIDs) == 0 {
+		return nil
+	}
+
+	tx, err := r.db.GetDB().Begin()
+	if err != nil {
+		return fmt.Errorf("failed to begin transaction: %w", err)
+	}
+	defer func() {
+		if err != nil {
+			tx.Rollback()
+		}
+	}()
+
+	query := `
+		UPDATE events
+		SET embedding_status = ?, updated_at = ?
+		WHERE id = ?
+	`
+
+	nowTime := time.Now()
+	stmt, err := tx.Prepare(query)
+	if err != nil {
+		return fmt.Errorf("failed to prepare statement: %w", err)
+	}
+	defer stmt.Close()
+
+	for _, id := range eventIDs {
+		_, err = stmt.Exec(status, nowTime, id)
+		if err != nil {
+			r.logger.Error("Failed to update event status for ID %d: %v", id, err)
+			return fmt.Errorf("failed to update event status for ID %d: %w", id, err)
+		}
+	}
+
+	if err = tx.Commit(); err != nil {
+		return fmt.Errorf("failed to commit transaction: %w", err)
+	}
+
+	r.logger.Info("Successfully updated status for %d events", len(eventIDs))
+	return nil
+}
+
 // GetExpiredEventIDs 获取过期事件的ID列表
 func (r *eventRepository) GetExpiredEventIDs(cutoffTime time.Time) ([]int64, error) {
 	query := `