|
| 1 | +package chipingressset |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "encoding/json" |
| 6 | + "fmt" |
| 7 | + "maps" |
| 8 | + "os" |
| 9 | + "path" |
| 10 | + "path/filepath" |
| 11 | + "slices" |
| 12 | + "strings" |
| 13 | + |
| 14 | + "github.com/jhump/protocompile" |
| 15 | + "google.golang.org/protobuf/reflect/protoreflect" |
| 16 | + |
| 17 | + cc "github.com/smartcontractkit/atlas/chip-config/client" // TODO: can we move it to chainlink-common? |
| 18 | + "github.com/smartcontractkit/chainlink-common/pkg/chipingress/pb" |
| 19 | +) |
| 20 | + |
| 21 | +// code copied from: https://github.com/smartcontractkit/atlas/blob/master/chip-cli/config/config.go and https://github.com/smartcontractkit/atlas/blob/master/chip-cli/config/proto_validator.go |
| 22 | +// reason: avoid dependency on the chip-cli module in the testing framework |
| 23 | +func chipConfigClient(ctx context.Context, chipConfigOutput *ChipConfigOutput) (cc.ChipConfigClient, error) { |
| 24 | + fmt.Printf("🔌 Initiating connection to Chip Config at \033[1m%s\033[0m...\n\n", chipConfigOutput.GRPCExternalURL) |
| 25 | + |
| 26 | + var clientOpts []cc.ClientOpt |
| 27 | + clientOpts = append(clientOpts, cc.WithBasicAuth(chipConfigOutput.Username, chipConfigOutput.Password)) |
| 28 | + |
| 29 | + client, err := cc.NewChipConfigClient(chipConfigOutput.GRPCExternalURL, clientOpts...) |
| 30 | + if err != nil { |
| 31 | + return nil, fmt.Errorf("failed to create Chip Config client: %w", err) |
| 32 | + } |
| 33 | + |
| 34 | + // Check we can connect to the server |
| 35 | + _, pErr := client.Ping(ctx) |
| 36 | + if pErr != nil { |
| 37 | + return nil, fmt.Errorf("failed to connect to Chip Config: %w", pErr) |
| 38 | + } |
| 39 | + |
| 40 | + fmt.Printf("🔗 Connected to Chip Config\n\n") |
| 41 | + |
| 42 | + return client, nil |
| 43 | +} |
| 44 | + |
| 45 | +func convertToPbSchemas(schemas map[string]*Schema, domain string) []*pb.Schema { |
| 46 | + pbSchemas := make([]*pb.Schema, len(schemas)) |
| 47 | + |
| 48 | + for i, schema := range slices.Collect(maps.Values(schemas)) { |
| 49 | + |
| 50 | + pbReferences := make([]*pb.SchemaReference, len(schema.References)) |
| 51 | + for j, reference := range schema.References { |
| 52 | + pbReferences[j] = &pb.SchemaReference{ |
| 53 | + Subject: fmt.Sprintf("%s-%s", domain, reference.Entity), |
| 54 | + Name: reference.Name, |
| 55 | + // Explicitly omit Version, this tells chip-config to use the latest version of the schema for this reference |
| 56 | + } |
| 57 | + } |
| 58 | + |
| 59 | + pbSchema := &pb.Schema{ |
| 60 | + Subject: fmt.Sprintf("%s-%s", domain, schema.Entity), |
| 61 | + Schema: schema.SchemaContent, |
| 62 | + References: pbReferences, |
| 63 | + } |
| 64 | + |
| 65 | + // If the schema has metadata, we need to add pb metadata to the schema |
| 66 | + if schema.Metadata.Stores != nil { |
| 67 | + |
| 68 | + stores := make(map[string]*pb.Store, len(schema.Metadata.Stores)) |
| 69 | + for key, store := range schema.Metadata.Stores { |
| 70 | + stores[key] = &pb.Store{ |
| 71 | + Index: store.Index, |
| 72 | + Partition: store.Partition, |
| 73 | + } |
| 74 | + } |
| 75 | + |
| 76 | + pbSchema.Metadata = &pb.MetaData{ |
| 77 | + Stores: stores, |
| 78 | + } |
| 79 | + } |
| 80 | + |
| 81 | + pbSchemas[i] = pbSchema |
| 82 | + } |
| 83 | + |
| 84 | + return pbSchemas |
| 85 | +} |
| 86 | + |
| 87 | +type RegistrationConfig struct { |
| 88 | + Domain string `json:"domain"` |
| 89 | + Schemas []Schema `json:"schemas"` |
| 90 | +} |
| 91 | + |
| 92 | +type Schema struct { |
| 93 | + Entity string `json:"entity"` |
| 94 | + Path string `json:"path"` |
| 95 | + References []SchemaReference `json:"references,omitempty"` |
| 96 | + SchemaContent string |
| 97 | + Metadata Metadata `json:"metadata,omitempty"` |
| 98 | +} |
| 99 | + |
| 100 | +type Metadata struct { |
| 101 | + Stores map[string]Store `json:"stores"` |
| 102 | +} |
| 103 | + |
| 104 | +type Store struct { |
| 105 | + Index []string `json:"index"` |
| 106 | + Partition []string `json:"partition"` |
| 107 | +} |
| 108 | + |
| 109 | +type SchemaReference struct { |
| 110 | + Name string `json:"name"` |
| 111 | + Entity string `json:"entity"` |
| 112 | + Path string `json:"path"` |
| 113 | +} |
| 114 | + |
| 115 | +func parseSchemaConfig(configFilePath, schemaDir string) (*RegistrationConfig, map[string]*Schema, error) { |
| 116 | + cfg, err := readConfig(configFilePath) |
| 117 | + if err != nil { |
| 118 | + return nil, nil, err |
| 119 | + } |
| 120 | + |
| 121 | + if err := ValidateEntityNames(cfg, schemaDir); err != nil { |
| 122 | + return nil, nil, fmt.Errorf("entity name validation failed: %w", err) |
| 123 | + } |
| 124 | + |
| 125 | + // Our end goal is to generate a schema registration request to chip config |
| 126 | + // We will use a map to store the schemas by entity and path |
| 127 | + // this is because more than one schema may reference the same schema |
| 128 | + // technically, since SR is idempotent, this is not strictly necessary, as duplicate registrations are noop |
| 129 | + schemas := make(map[string]*Schema) |
| 130 | + |
| 131 | + for _, schema := range cfg.Schemas { |
| 132 | + |
| 133 | + // For each of the schemas, we need to get the references schema content |
| 134 | + for _, reference := range schema.References { |
| 135 | + |
| 136 | + // read schema contents |
| 137 | + refSchemaContent, err := os.ReadFile(path.Join(schemaDir, reference.Path)) |
| 138 | + if err != nil { |
| 139 | + return nil, nil, fmt.Errorf("error reading schema: %v", err) |
| 140 | + } |
| 141 | + |
| 142 | + // generate key with entity and path since other schemas may also reference this schema |
| 143 | + key := fmt.Sprintf("%s:%s", reference.Entity, reference.Path) |
| 144 | + |
| 145 | + // if the schema already exists, skip it |
| 146 | + if _, ok := schemas[key]; ok { |
| 147 | + continue |
| 148 | + } |
| 149 | + |
| 150 | + schemas[key] = &Schema{ |
| 151 | + Entity: reference.Entity, |
| 152 | + Path: reference.Path, |
| 153 | + SchemaContent: string(refSchemaContent), |
| 154 | + } |
| 155 | + } |
| 156 | + |
| 157 | + // add the root schema to the map |
| 158 | + schemaContent, err := os.ReadFile(path.Join(schemaDir, schema.Path)) |
| 159 | + if err != nil { |
| 160 | + return nil, nil, fmt.Errorf("error reading schema: %v", err) |
| 161 | + } |
| 162 | + |
| 163 | + key := fmt.Sprintf("%s:%s", schema.Entity, schema.Path) |
| 164 | + // if the schema already exists, that means it is referenced by another schema. |
| 165 | + // so we just need to add the references to the existing schema in the map |
| 166 | + if s, ok := schemas[key]; ok { |
| 167 | + s.References = append(s.References, schema.References...) |
| 168 | + continue |
| 169 | + } |
| 170 | + |
| 171 | + schemas[key] = &Schema{ |
| 172 | + Entity: schema.Entity, |
| 173 | + Path: schema.Path, |
| 174 | + SchemaContent: string(schemaContent), |
| 175 | + References: schema.References, |
| 176 | + } |
| 177 | + |
| 178 | + } |
| 179 | + |
| 180 | + return cfg, schemas, nil |
| 181 | +} |
| 182 | + |
| 183 | +func readConfig(path string) (*RegistrationConfig, error) { |
| 184 | + f, err := os.Open(path) |
| 185 | + if err != nil { |
| 186 | + return nil, fmt.Errorf("failed to open config file '%s': %w", path, err) |
| 187 | + } |
| 188 | + defer f.Close() |
| 189 | + |
| 190 | + var cfg RegistrationConfig |
| 191 | + |
| 192 | + dErr := json.NewDecoder(f).Decode(&cfg) |
| 193 | + if dErr != nil { |
| 194 | + return nil, fmt.Errorf("failed to decode config: %w", dErr) |
| 195 | + } |
| 196 | + |
| 197 | + return &cfg, nil |
| 198 | +} |
| 199 | + |
| 200 | +// ValidateEntityNames validates that all entity names in the config match the fully qualified |
| 201 | +// protobuf names (package.MessageName) from their corresponding proto files. |
| 202 | +// It collects all validation errors and returns them together for better user experience. |
| 203 | +func ValidateEntityNames(cfg *RegistrationConfig, schemaDir string) error { |
| 204 | + var errors []string |
| 205 | + |
| 206 | + for _, schema := range cfg.Schemas { |
| 207 | + if err := validateEntityName(schema.Entity, schema.Path, schemaDir); err != nil { |
| 208 | + errors = append(errors, fmt.Sprintf(" - schema '%s': %s", schema.Path, err)) |
| 209 | + } |
| 210 | + |
| 211 | + for _, ref := range schema.References { |
| 212 | + if err := validateEntityName(ref.Entity, ref.Path, schemaDir); err != nil { |
| 213 | + errors = append(errors, fmt.Sprintf(" - referenced schema '%s': %s", ref.Path, err)) |
| 214 | + } |
| 215 | + } |
| 216 | + } |
| 217 | + |
| 218 | + if len(errors) > 0 { |
| 219 | + return fmt.Errorf("entity name validation failed with %d error(s):\n%s", len(errors), strings.Join(errors, "\n")) |
| 220 | + } |
| 221 | + |
| 222 | + return nil |
| 223 | +} |
| 224 | + |
| 225 | +func validateEntityName(entityName, protoPath, schemaDir string) error { |
| 226 | + fullPath := path.Join(schemaDir, protoPath) |
| 227 | + |
| 228 | + // Find the message descriptor that matches the entity name |
| 229 | + msgDesc, err := findMessageDescriptor(fullPath, entityName) |
| 230 | + if err != nil { |
| 231 | + return fmt.Errorf("failed to find message descriptor in '%s': %w", protoPath, err) |
| 232 | + } |
| 233 | + |
| 234 | + // Extract the expected entity name from the message descriptor |
| 235 | + expectedEntity := string(msgDesc.FullName()) |
| 236 | + if entityName != expectedEntity { |
| 237 | + return fmt.Errorf( |
| 238 | + "entity name mismatch in chip.json:\n"+ |
| 239 | + " Proto file: %s\n"+ |
| 240 | + " Expected: %s\n"+ |
| 241 | + " Got: %s\n"+ |
| 242 | + " \n"+ |
| 243 | + " The entity name must be the fully qualified protobuf name: {package}.{MessageName}", |
| 244 | + protoPath, |
| 245 | + expectedEntity, |
| 246 | + entityName, |
| 247 | + ) |
| 248 | + } |
| 249 | + |
| 250 | + return nil |
| 251 | +} |
| 252 | + |
| 253 | +// findMessageDescriptor finds a message descriptor by name (either full name or short name) |
| 254 | +// This matches the logic in chip-ingress/internal/serde/message.go |
| 255 | +func findMessageDescriptor(filePath, targetMessageName string) (protoreflect.MessageDescriptor, error) { |
| 256 | + compiler := protocompile.Compiler{ |
| 257 | + Resolver: &protocompile.SourceResolver{ |
| 258 | + ImportPaths: getImportPaths(filePath, 3), |
| 259 | + }, |
| 260 | + } |
| 261 | + |
| 262 | + filename := filepath.Base(filePath) |
| 263 | + fds, err := compiler.Compile(context.Background(), filename) |
| 264 | + if err != nil { |
| 265 | + return nil, fmt.Errorf("failed to compile proto file: %w", err) |
| 266 | + } |
| 267 | + |
| 268 | + if len(fds) == 0 { |
| 269 | + return nil, fmt.Errorf("no file descriptors found") |
| 270 | + } |
| 271 | + |
| 272 | + // Search through all file descriptors for the target message |
| 273 | + for _, fd := range fds { |
| 274 | + messages := fd.Messages() |
| 275 | + for i := range messages.Len() { |
| 276 | + msgDesc := messages.Get(i) |
| 277 | + |
| 278 | + // Match by full name (e.g., "package.MessageName") or short name (e.g., "MessageName") |
| 279 | + if string(msgDesc.FullName()) == targetMessageName || string(msgDesc.Name()) == targetMessageName { |
| 280 | + return msgDesc, nil |
| 281 | + } |
| 282 | + } |
| 283 | + } |
| 284 | + |
| 285 | + return nil, fmt.Errorf("message descriptor not found for name: %s", targetMessageName) |
| 286 | +} |
| 287 | + |
| 288 | +func getImportPaths(path string, depth int) []string { |
| 289 | + paths := make([]string, 0, depth+1) |
| 290 | + paths = append(paths, filepath.Dir(path)) |
| 291 | + |
| 292 | + currentPath := path |
| 293 | + for i := 0; i < depth; i++ { |
| 294 | + currentPath = filepath.Dir(currentPath) |
| 295 | + paths = append(paths, currentPath) |
| 296 | + } |
| 297 | + return paths |
| 298 | +} |
0 commit comments