Add CHASM Visibility registration and task processing

chasm/lib/tests/library.go

@@ -16,7 +16,8 @@ func (l *library) Name() string {
 
 func (l *library) Components() []*chasm.RegistrableComponent {
 	return []*chasm.RegistrableComponent{
-		chasm.NewRegistrableComponent[*PayloadStore]("payloadStore"),
+		chasm.NewRegistrableComponent[*PayloadStore]("payloadStore",
+			chasm.WithSearchAttributes(TestKeywordSearchAttribute)),
 	}
 }
 

chasm/lib/tests/payload.go

@@ -24,6 +24,8 @@ const (
 )
 
 var (
+	TestKeywordSearchAttribute = chasm.NewSearchAttributeKeyword(TestKeywordSAFieldName, chasm.SearchAttributeFieldKeyword01)
+
 	_ chasm.VisibilitySearchAttributesProvider = (*PayloadStore)(nil)
 	_ chasm.VisibilityMemoProvider             = (*PayloadStore)(nil)
 )
@@ -147,12 +149,10 @@ func (s *PayloadStore) LifecycleState(
 
 // SearchAttributes implements chasm.VisibilitySearchAttributesProvider interface
 func (s *PayloadStore) SearchAttributes(
-	_ chasm.Context,
-) map[string]chasm.VisibilityValue {
-	// TODO: UpsertSearchAttribute as well when CHASM framework supports Per-Component SearchAttributes
-	// For now, we just update a random existing pre-defined SA to make sure the logic works.
-	return map[string]chasm.VisibilityValue{
-		TestKeywordSAFieldName: chasm.VisibilityValueString(TestKeywordSAFieldValue),
+	ctx chasm.Context,
+) []chasm.SearchAttributeKeyValue {
+	return []chasm.SearchAttributeKeyValue{
+		TestKeywordSearchAttribute.Value(TestKeywordSAFieldValue),
 	}
 }
 

chasm/registrable_component.go

@@ -1,9 +1,15 @@
 package chasm
 
 import (
+	"fmt"
 	"reflect"
+
+	enumspb "go.temporal.io/api/enums/v1"
+	"go.temporal.io/api/serviceerror"
 )
 
+var _ VisibilitySearchAttributesMapper = (*RegistrableComponent)(nil)
+
 type (
 	RegistrableComponent struct {
 		componentType string
@@ -13,6 +19,11 @@ type (
 		ephemeral     bool
 		singleCluster bool
 		shardingFn    func(EntityKey) string
+
+		// Search attribute mappings
+		aliasToField map[string]string
+		fieldToAlias map[string]string
+		saTypeMap    map[string]enumspb.IndexedValueType
 	}
 
 	RegistrableComponentOption func(*RegistrableComponent)
@@ -56,6 +67,44 @@ func WithShardingFn(
 	}
 }
 
+func WithSearchAttributes(
+	searchAttributes ...SearchAttribute,
+) RegistrableComponentOption {
+	return func(rc *RegistrableComponent) {
+		if len(searchAttributes) == 0 {
+			return
+		}
+		rc.aliasToField = make(map[string]string, len(searchAttributes))
+		rc.fieldToAlias = make(map[string]string, len(searchAttributes))
+		rc.saTypeMap = make(map[string]enumspb.IndexedValueType, len(searchAttributes))
+
+		for _, sa := range searchAttributes {
+			alias := sa.definition().Alias
+			field := sa.definition().Field
+			valueType := sa.definition().ValueType
+
+			rc.aliasToField[alias] = field
+			rc.fieldToAlias[field] = alias
+			rc.saTypeMap[field] = valueType
+		}
+	}
+}
+
+// validate checks for errors in the component configuration.
+func (rc *RegistrableComponent) validate() error {
+	// Check for duplicate field names in search attributes
+	fieldToAliases := make(map[string][]string)
+	for alias, field := range rc.aliasToField {
+		fieldToAliases[field] = append(fieldToAliases[field], alias)
+	}
+	for field, aliases := range fieldToAliases {
+		if len(aliases) > 1 {
+			return fmt.Errorf("search attributes contain duplicate field names: field '%s' is used by multiple aliases: %v", field, aliases)
+		}
+	}
+	return nil
+}
+
 // fqType returns the fully qualified name of the component, which is a combination of
 // the library name and the component type. This is used to uniquely identify
 // the component in the registry.
@@ -66,3 +115,25 @@ func (rc RegistrableComponent) fqType() string {
 	}
 	return fullyQualifiedName(rc.library.Name(), rc.componentType)
 }
+
+// SearchAttributeAlias returns the search attribute alias for the given field name.
+func (rc *RegistrableComponent) SearchAttributeAlias(field string) (string, error) {
+	alias, ok := rc.fieldToAlias[field]
+	if !ok {
+		return "", serviceerror.NewInvalidArgument(
+			fmt.Sprintf("registrable component name %s has no alias defined for field name %s", rc.fqType(), field),
+		)
+	}
+	return alias, nil
+}
+
+// SearchAttributeField returns the search attribute field name for the given alias.
+func (rc *RegistrableComponent) SearchAttributeField(alias string) (string, error) {
+	field, ok := rc.aliasToField[alias]
+	if !ok {
+		return "", serviceerror.NewInvalidArgument(
+			fmt.Sprintf("registrable component name %s has no field defined for alias name %s", rc.fqType(), alias),
+		)
+	}
+	return field, nil
+}

chasm/registry.go

@@ -106,6 +106,9 @@ func (r *Registry) registerComponent(
 	if err := r.validateName(rc.componentType); err != nil {
 		return err
 	}
+	if err := rc.validate(); err != nil {
+		return err
+	}
 	fqn := fullyQualifiedName(lib.Name(), rc.componentType)
 	if _, ok := r.componentByType[fqn]; ok {
 		return fmt.Errorf("component %s is already registered", fqn)