Method processImport

pkg/filemanager/workflows/import.go:123–177

(ctx context.Context, dep dependency.Dep)

}

func (m *ImportTask) processImport(ctx context.Context, dep dependency.Dep) (task.Status, error) {
	user := inventory.UserFromContext(ctx)

	dst, err := fs.NewUriFromString(m.state.Dst)
	if err != nil {
		return task.StatusError, fmt.Errorf("failed to parse dst: %s (%w)", err, queue.CriticalErr)
	}

	// Use a temporary file manager just for listing physical files
	listFm := manager.NewFileManager(dep, user)
	physicalFiles, err := listFm.ListPhysical(ctx, m.state.Src, m.state.PolicyID, m.state.Recursive,
		func(i int) {
			atomic.AddInt64(&m.progress[ProgressTypeIndexed].Current, int64(i))
		})
	listFm.Recycle()
	if err != nil {
		return task.StatusError, fmt.Errorf("failed to list physical files: %w", err)
	}

	m.l.Info("Importing %d physical files", len(physicalFiles))

	m.Lock()
	m.progress[ProgressTypeImported] = &queue.Progress{
		Total: int64(len(physicalFiles)),
	}
	delete(m.progress, ProgressTypeIndexed)
	m.Unlock()

	failed := 0
	totalFiles := len(physicalFiles)

	// Process files in batches to control memory usage
	for batchStart := 0; batchStart < totalFiles; batchStart += ImportBatchSize {
		batchEnd := min(batchStart+ImportBatchSize, totalFiles)

		batch := physicalFiles[batchStart:batchEnd]
		batchFailed := m.processBatch(ctx, dep, user, dst, batch)
		failed += batchFailed

		// Clear batch elements to allow GC of individual items
		for i := batchStart; i < batchEnd; i++ {
			physicalFiles[i] = fs.PhysicalObject{}
		}

		// Run GC after each batch to free memory
		runtime.GC()
	}

	// Clear the entire slice to allow GC
	physicalFiles = nil
	runtime.GC()

	m.state.Failed = failed
	return task.StatusCompleted, nil
}

// processBatch processes a batch of physical files with a fresh file manager.
func (m *ImportTask) processBatch(ctx context.Context, dep dependency.Dep, user *ent.User, dst *fs.URI, batch []fs.PhysicalObject) int {