fix: updated graphQL node implementation (#365)

* Updated graphQL node implementation

* Fixed execution output of deployed workflow for ContractWRite

* Make sure metadata uses original address but not tokenContract

* Fixed variables shared across iteration in parallel execution from LoopNode

Author: will-dz

core/taskengine/run_node_immediately.go

@@ -72,7 +72,14 @@ func (n *Engine) runBlockTriggerImmediately(triggerConfig map[string]interface{}
 
 	// Check if a specific block number is requested
 	if configBlockNumber, ok := triggerConfig["blockNumber"]; ok {
-		if blockNum, err := n.parseUint64(configBlockNumber); err == nil {
+		blockNum, err := n.parseUint64(configBlockNumber)
+		if err != nil {
+			if n.logger != nil {
+				n.logger.Debug("Failed to parse blockNumber from trigger config, using latest block",
+					"blockNumber", configBlockNumber,
+					"error", err)
+			}
+		} else {
 			blockNumber = blockNum
 		}
 	}
@@ -351,7 +358,7 @@ func (n *Engine) runEventTriggerWithTenderlySimulation(ctx context.Context, quer
 	}
 
 	metadata := map[string]interface{}{
-		"tokenContract":    simulatedLog.Address.Hex(), // Renamed for clarity
+		"address":          simulatedLog.Address.Hex(), // Original contract address
 		"topics":           topicsMetadata,             // Now protobuf-compatible
 		"data":             "0x" + common.Bytes2Hex(simulatedLog.Data),
 		"blockNumber":      simulatedLog.BlockNumber,
@@ -1080,7 +1087,7 @@ func (n *Engine) runEventTriggerWithHistoricalSearch(ctx context.Context, querie
 
 	// Build raw metadata (the original blockchain event data)
 	metadata := map[string]interface{}{
-		"tokenContract":    mostRecentEvent.Address.Hex(), // Renamed for clarity
+		"address":          mostRecentEvent.Address.Hex(), // Original contract address
 		"topics":           topics,
 		"data":             "0x" + common.Bytes2Hex(mostRecentEvent.Data),
 		"blockNumber":      mostRecentEvent.BlockNumber,
@@ -1501,7 +1508,7 @@ func (n *Engine) runProcessingNodeWithInputs(nodeType string, nodeConfig map[str
 		secrets = make(map[string]string)
 	}
 
-	// Create a clean VM for isolated execution with proper secrets
+	// Create a clean VM for isolated execution with proper secrets (no task needed for immediate execution)
 	vm, err := NewVMWithData(nil, nil, n.smartWalletConfig, secrets)
 	if err != nil {
 		return nil, fmt.Errorf("failed to create VM: %w", err)
@@ -1679,16 +1686,32 @@ func (n *Engine) extractExecutionResult(executionStep *avsproto.Execution_Step)
 		result["success"] = true
 	} else if contractWrite := executionStep.GetContractWrite(); contractWrite != nil {
 		// ContractWrite output now contains enhanced results structure
+		// Data contains decoded events (flattened by method name), Metadata contains method results
+
+		// Extract data if available (decoded events organized by method name)
 		if contractWrite.GetData() != nil {
-			// Extract results using helper function
-			allResults := ExtractResultsFromProtobufValue(contractWrite.GetData())
+			iface := contractWrite.GetData().AsInterface()
+			if m, ok := iface.(map[string]interface{}); ok {
+				result["data"] = m
+			} else {
+				result["data"] = iface
+			}
+		}
 
-			// Return results array directly without backward compatibility
+		// Extract metadata if available (method results array)
+		if contractWrite.GetMetadata() != nil {
+			// Extract results from metadata (method results array)
+			allResults := ExtractResultsFromProtobufValue(contractWrite.GetMetadata())
 			result["results"] = allResults
 
-			return result, nil
+			if metadataArray := gow.ValueToSlice(contractWrite.GetMetadata()); metadataArray != nil {
+				result["metadata"] = metadataArray
+			} else {
+				result["metadata"] = contractWrite.GetMetadata().AsInterface()
+			}
 		}
-		return map[string]interface{}{"status": "success"}, nil
+
+		return result, nil
 	} else if loop := executionStep.GetLoop(); loop != nil {
 		// Loop output contains the array of iteration results
 		if loop.GetData() != nil {
@@ -1709,6 +1732,19 @@ func (n *Engine) extractExecutionResult(executionStep *avsproto.Execution_Step)
 				result["data"] = iface
 			}
 		}
+	} else if graphql := executionStep.GetGraphql(); graphql != nil {
+		// GraphQL output contains the query results
+		if graphql.GetData() != nil {
+			// Extract the actual GraphQL response data
+			iface := graphql.GetData().AsInterface()
+			// Return the GraphQL data directly (it should already be in the correct format)
+			if graphqlData, ok := iface.(map[string]interface{}); ok {
+				return graphqlData, nil
+			} else {
+				// Fallback: wrap in data field
+				result["data"] = iface
+			}
+		}
 	}
 
 	// If no specific data was extracted, include basic execution info
@@ -1976,10 +2012,15 @@ func (n *Engine) RunNodeImmediatelyRPC(user *model.User, req *avsproto.RunNodeWi
 	case NodeTypeContractWrite:
 		// For contract write nodes - always set output structure to avoid OUTPUT_DATA_NOT_SET
 		contractWriteOutput := &avsproto.ContractWriteNode_Output{}
+		// Always create the data structure, even if result is nil/empty
+		var resultsArray []interface{}
+		var decodedEventsData = make(map[string]interface{})
+
 		if result != nil && len(result) > 0 {
-			// Convert result to the new data structure
-			var resultsArray []interface{}
-			var decodedEventsData = make(map[string]interface{})
+			// First, try to extract data directly (flattened events organized by method name)
+			if dataFromVM, ok := result["data"].(map[string]interface{}); ok {
+				decodedEventsData = dataFromVM
+			}
 
 			// Check if we have the new results array format (from VM execution)
 			if resultsFromVM, ok := result["results"].([]interface{}); ok {
@@ -2012,12 +2053,13 @@ func (n *Engine) RunNodeImmediatelyRPC(user *model.User, req *avsproto.RunNodeWi
 
 						resultsArray = append(resultsArray, convertedResult)
 
-						// 🚀 NEW: Decode event logs from transaction receipt
+						// 🚀 NEW: Parse events for this specific method and store by method name
+						methodEvents := make(map[string]interface{})
 						if methodResult.Receipt != nil {
 							receiptData := methodResult.Receipt.AsInterface()
 							if receiptMap, ok := receiptData.(map[string]interface{}); ok {
 								if logs, hasLogs := receiptMap["logs"]; hasLogs {
-									if logsArray, ok := logs.([]interface{}); ok {
+									if logsArray, ok := logs.([]interface{}); ok && len(logsArray) > 0 {
 										// Get contract ABI for event decoding
 										var contractABI *abi.ABI
 										if methodResult.MethodAbi != nil {
@@ -2034,7 +2076,7 @@ func (n *Engine) RunNodeImmediatelyRPC(user *model.User, req *avsproto.RunNodeWi
 											}
 										}
 
-										// Decode each event log
+										// Decode each event log for this method
 										for _, logInterface := range logsArray {
 											if logMap, ok := logInterface.(map[string]interface{}); ok {
 												if contractABI != nil {
@@ -2070,39 +2112,12 @@ func (n *Engine) RunNodeImmediatelyRPC(user *model.User, req *avsproto.RunNodeWi
 
 													// Decode the event using existing logic
 													if decodedEvent, err := n.parseEventWithParsedABI(eventLog, contractABI, nil); err == nil {
-														// Add decoded event data to flattened result
+														// Flatten event fields into methodEvents
 														for key, value := range decodedEvent {
-															// Use eventName.fieldName format with dot notation to avoid conflicts
-															if eventName, hasEventName := decodedEvent["eventName"]; hasEventName {
-																if eventNameStr, ok := eventName.(string); ok && key != "eventName" {
-																	flatKey := fmt.Sprintf("%s.%s", eventNameStr, key)
-																	decodedEventsData[flatKey] = value
-																}
+															if key != "eventName" { // Skip meta field
+																methodEvents[key] = value
 															}
 														}
-
-														// Also add the complete event object with a descriptive key for multiple events
-														var eventKey string
-														if eventName, hasEventName := decodedEvent["eventName"]; hasEventName {
-															if eventNameStr, ok := eventName.(string); ok {
-																// Try to get transaction hash or log index for uniqueness
-																var uniqueSuffix string
-																if txHash, hasTxHash := logMap["transactionHash"]; hasTxHash {
-																	if txHashStr, ok := txHash.(string); ok {
-																		uniqueSuffix = txHashStr
-																	}
-																} else if logIndex, hasLogIndex := logMap["logIndex"]; hasLogIndex {
-																	uniqueSuffix = fmt.Sprintf("%x", logIndex)
-																} else {
-																	// Fallback to eventLog.Index if logMap doesn't have logIndex
-																	uniqueSuffix = fmt.Sprintf("%d", eventLog.Index)
-																}
-																eventKey = fmt.Sprintf("%s_%s", eventNameStr, uniqueSuffix)
-															}
-														} else {
-															eventKey = fmt.Sprintf("event_%d", eventLog.Index)
-														}
-														decodedEventsData[eventKey] = decodedEvent
 													}
 												}
 											}
@@ -2111,6 +2126,9 @@ func (n *Engine) RunNodeImmediatelyRPC(user *model.User, req *avsproto.RunNodeWi
 								}
 							}
 						}
+
+						// Store events for this method (empty object if no events)
+						decodedEventsData[methodResult.MethodName] = methodEvents
 					} else if methodResultMap, ok := resultInterface.(map[string]interface{}); ok {
 						// Already in map format
 						resultsArray = append(resultsArray, methodResultMap)
@@ -2139,18 +2157,18 @@ func (n *Engine) RunNodeImmediatelyRPC(user *model.User, req *avsproto.RunNodeWi
 				}
 			}
 
-			// 🚀 NEW: Set data field with decoded events (flattened format like ContractRead)
-			if len(decodedEventsData) > 0 {
-				if dataValue, err := structpb.NewValue(decodedEventsData); err == nil {
-					contractWriteOutput.Data = dataValue
-				}
-			}
+		}
 
-			// 🚀 NEW: Set metadata field with detailed method information
-			if len(resultsArray) > 0 {
-				if metadataValue, err := structpb.NewValue(resultsArray); err == nil {
-					contractWriteOutput.Metadata = metadataValue
-				}
+		// 🚀 NEW: Set data field with decoded events (flattened format like ContractRead)
+		// Always set data field - empty object if no events, flattened events if present
+		if dataValue, err := structpb.NewValue(decodedEventsData); err == nil {
+			contractWriteOutput.Data = dataValue
+		}
+
+		// 🚀 NEW: Set metadata field with detailed method information
+		if len(resultsArray) > 0 {
+			if metadataValue, err := structpb.NewValue(resultsArray); err == nil {
+				contractWriteOutput.Metadata = metadataValue
 			}
 		}
 		resp.OutputData = &avsproto.RunNodeWithInputsResp_ContractWrite{

core/taskengine/shared_event_enrichment.go

@@ -138,6 +138,9 @@ func parseEventWithABIShared(eventLog *types.Log, contractABI *abi.ABI, query *a
 	parsedData := make(map[string]interface{})
 	parsedData["eventName"] = eventName
 
+	// Create ABI value converter for proper type conversion
+	converter := NewABIValueConverter(nil, nil)
+
 	// Add indexed parameters from topics (skip topic[0] which is event signature)
 	indexedCount := 0
 	nonIndexedCount := 0
@@ -151,14 +154,17 @@ func parseEventWithABIShared(eventLog *types.Log, contractABI *abi.ABI, query *a
 					// Convert hash to address for indexed address parameters
 					parsedData[input.Name] = common.HexToAddress(eventLog.Topics[topicIndex].Hex()).Hex()
 				} else {
-					parsedData[input.Name] = eventLog.Topics[topicIndex]
+					// Convert common.Hash to hex string for structpb compatibility
+					parsedData[input.Name] = eventLog.Topics[topicIndex].Hex()
 				}
 			}
 			indexedCount++
 		} else {
-			// Get from decoded data
+			// Get from decoded data and convert to structpb-compatible type
 			if nonIndexedCount < len(decodedData) {
-				parsedData[input.Name] = decodedData[nonIndexedCount]
+				rawValue := decodedData[nonIndexedCount]
+				// Use ABI converter to ensure structpb compatibility
+				parsedData[input.Name] = converter.ConvertABIValueToInterface(rawValue, input.Type, input.Name)
 			}
 			nonIndexedCount++
 		}
@@ -368,13 +374,19 @@ func findCommonAddress(fromAddresses, toAddresses []string) string {
 
 // createBasicEventMetadata creates basic metadata structure for events without ABI
 func createBasicEventMetadata(eventLog *types.Log) map[string]interface{} {
+	// Convert topics from []common.Hash to []string for structpb compatibility
+	topicsHex := make([]string, len(eventLog.Topics))
+	for i, topic := range eventLog.Topics {
+		topicsHex[i] = topic.Hex()
+	}
+
 	return map[string]interface{}{
 		"tokenContract":   eventLog.Address.Hex(), // Renamed from "address" for clarity
 		"blockNumber":     eventLog.BlockNumber,
 		"transactionHash": eventLog.TxHash.Hex(),
 		"logIndex":        eventLog.Index,
 		"blockHash":       eventLog.BlockHash.Hex(),
-		"topics":          eventLog.Topics,
+		"topics":          topicsHex,
 		"data":            common.Bytes2Hex(eventLog.Data),
 		"removed":         eventLog.Removed,
 	}

core/taskengine/tenderly_client.go

@@ -50,6 +50,7 @@ type RPCError struct {
 // Call parameters for eth_call
 type CallParams struct {
 	To   string `json:"to"`
+	From string `json:"from,omitempty"`
 	Data string `json:"data"`
 }
 
@@ -538,16 +539,18 @@ func (tc *TenderlyClient) createMockTransferLog(contractAddress string, from, to
 }
 
 // SimulateContractWrite simulates a contract write operation using Tenderly
-func (tc *TenderlyClient) SimulateContractWrite(ctx context.Context, contractAddress string, callData string, contractABI string, methodName string, chainID int64) (*ContractWriteSimulationResult, error) {
+func (tc *TenderlyClient) SimulateContractWrite(ctx context.Context, contractAddress string, callData string, contractABI string, methodName string, chainID int64, fromAddress string) (*ContractWriteSimulationResult, error) {
 	tc.logger.Info("🔮 Simulating contract write via Tenderly",
 		"contract", contractAddress,
 		"method", methodName,
+		"from", fromAddress,
 		"chain_id", chainID)
 
 	// For simulation, we use eth_call to see what would happen without actually executing
 	// This gives us the return data and potential revert reasons
 	callParams := CallParams{
 		To:   contractAddress,
+		From: fromAddress,
 		Data: callData,
 	}
 
@@ -612,7 +615,7 @@ func (tc *TenderlyClient) SimulateContractWrite(ctx context.Context, contractAdd
 	result.Transaction = &ContractWriteTransactionData{
 		Hash:       fmt.Sprintf("0x%064x", time.Now().UnixNano()), // Mock transaction hash
 		Status:     "simulated",
-		From:       "0x0000000000000000000000000000000000000001", // Mock sender
+		From:       fromAddress, // Use the actual user's wallet address
 		To:         contractAddress,
 		Value:      "0",
 		Timestamp:  time.Now().Unix(),

core/taskengine/tenderly_client_test.go

@@ -1516,7 +1516,7 @@ func TestEventTriggerImmediately_TenderlySimulation_Unit(t *testing.T) {
 		require.True(t, ok, "metadata should be a map[string]interface{}")
 		require.NotNil(t, metadata, "Should have metadata")
 
-		assert.NotNil(t, metadata["tokenContract"], "Should have address in metadata")
+		assert.NotNil(t, metadata["address"], "Should have address in metadata")
 		assert.NotNil(t, metadata["blockNumber"], "Should have blockNumber in metadata")
 
 		t.Logf("✅ Tenderly simulation successful!")