backward compatible metadata keys (#4738)

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

- **New Features**
- Introduced a new constant for header prefix to improve metadata
handling.
  
- **Improvements**
- Enhanced key retrieval logic for better backward compatibility in
metadata access.
- Improved error logging in message handling to include subscriber name
for better traceability.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com>

Author: wdbaruni

pkg/lib/envelope/metadata.go

@@ -2,11 +2,13 @@ package envelope
 
 import (
 	"strconv"
+	"strings"
 	"time"
 )
 
 // Metadata keys
 const (
+	HeaderPrefix       = "Bacalhau-"
 	KeyMessageType     = "Bacalhau-Type"
 	KeyPayloadEncoding = "Bacalhau-PayloadEncoding"
 )
@@ -50,7 +52,15 @@ func NewMetadataFromMapCopy(m map[string]string) *Metadata {
 
 // Get returns the value for a given key, or an empty string if the key doesn't exist
 func (m Metadata) Get(key string) string {
-	return m[key]
+	if v, ok := m[key]; ok {
+		return v
+	}
+	// backward compatible with old headers
+	if key == KeyMessageType || key == KeyPayloadEncoding {
+		// return keys excluding header prefix
+		return m[strings.TrimPrefix(key, HeaderPrefix)]
+	}
+	return ""
 }
 
 // Has checks if a key exists in the metadata

pkg/lib/ncl/subscriber.go

@@ -75,7 +75,7 @@ func (s *subscriber) Subscribe(ctx context.Context, subjects ...string) error {
 // messageHandler is the callback function for message processing
 func (s *subscriber) handleNatsMessage(m *nats.Msg) {
 	if err := s.processMessage(m); err != nil {
-		log.Error().Err(err).Msg("failed to process message")
+		log.Error().Err(err).Str("handler", s.config.Name).Msg("failed to process message")
 
 		s.consecutiveFailures += 1
 		delay := s.config.Backoff.BackoffDuration(s.consecutiveFailures)