Merge branch 'main' into LFXV2-1709

Author: prabodhcs

cmd/committee-api/main.go

@@ -106,6 +106,7 @@ func main() {
 		usecaseSvc.WithProjectRetriever(projectRetriever),
 		usecaseSvc.WithUserReader(userReader),
 		usecaseSvc.WithCommitteePublisher(committeePublisher),
+		usecaseSvc.WithB2BOrgResolver(service.B2BOrgResolverImpl(ctx)),
 	)
 
 	readCommitteeUseCase := usecaseSvc.NewCommitteeReaderOrchestrator(

cmd/committee-api/service/providers.go

@@ -31,10 +31,11 @@ import (
 )
 
 var (
-	natsStorage    port.CommitteeReaderWriter
-	natsMessaging  port.ProjectReader
-	natsUserReader port.UserReader
-	natsPublisher  port.CommitteePublisher
+	natsStorage        port.CommitteeReaderWriter
+	natsMessaging      port.ProjectReader
+	natsUserReader     port.UserReader
+	natsB2BOrgResolver port.B2BOrgResolver
+	natsPublisher      port.CommitteePublisher
 
 	// expose the NATS client for direct access in subscriptions
 	natsClient *nats.NATSClient
@@ -92,6 +93,7 @@ func natsInit(ctx context.Context) {
 		natsStorage = nats.NewStorage(client)
 		natsMessaging = nats.NewMessageRequest(client)
 		natsUserReader = nats.NewUserRequest(client)
+		natsB2BOrgResolver = nats.NewB2BOrgResolver(client)
 		natsPublisher = nats.NewMessagePublisher(client)
 	})
 }
@@ -231,6 +233,27 @@ func UserReaderImpl(ctx context.Context) port.UserReader {
 	return userReader
 }
 
+// B2BOrgResolverImpl initializes the b2b_org resolver used to validate committee member
+// organization.id values (LFXV2-2400). Returns nil in mock mode.
+func B2BOrgResolverImpl(ctx context.Context) port.B2BOrgResolver {
+	repoSource := os.Getenv("REPOSITORY_SOURCE")
+	if repoSource == "" {
+		repoSource = "nats"
+	}
+
+	switch repoSource {
+	case "mock":
+		return nil
+	case "nats":
+		natsInit(ctx)
+		return natsB2BOrgResolver
+	default:
+		log.Fatalf("unsupported b2b org resolver implementation: %s", repoSource)
+	}
+
+	return nil
+}
+
 // AuthServiceImpl initializes the authentication service implementation
 func AuthServiceImpl(ctx context.Context) port.Authenticator {
 	var authService port.Authenticator
@@ -852,6 +875,7 @@ func QueueSubscriptions(ctx context.Context, committeeReader port.CommitteeReade
 					usecaseSvc.WithProjectRetriever(ProjectRetrieverImpl(ctx)),
 					usecaseSvc.WithUserReader(UserReaderImpl(ctx)),
 					usecaseSvc.WithCommitteePublisher(CommitteePublisherImpl(ctx)),
+					usecaseSvc.WithB2BOrgResolver(B2BOrgResolverImpl(ctx)),
 				),
 			),
 			usecaseSvc.WithCommitteeWriterForMessageHandler(CommitteeWriterImpl(ctx)),

cmd/committee-cli/README.md

@@ -19,6 +19,9 @@ committee-cli <command> <subcommand> [subcommand flags]
 | Env var | Default | Description |
 |---|---|---|
 | `NATS_URL` | `nats://localhost:4222` | NATS server address |
+| `NATS_TIMEOUT` | `120s` | NATS connection dial/timeout (`nats.Timeout`); KV operations use caller context deadlines |
+| `OPENSEARCH_URL` | `http://localhost:9200` | OpenSearch base URL (`sync member-cdp-org-id`) |
+| `OPENSEARCH_INDEX` | `resources` | OpenSearch resources index (`sync member-cdp-org-id`) |
 | `LOG_LEVEL` | `debug` | Log verbosity (e.g. `info`) |
 
 ### Commands
@@ -136,6 +139,42 @@ NATS_URL=nats://localhost:4222 \
   committee-cli sync reindex-invites --committee-uid=abc-123
 ```
 
+#### `sync member-cdp-org-id`
+
+Repairs committee members that store a **CDP organization UUID** in `organization.id` (self-serve PR #779). Discovers affected members via OpenSearch (`committee_member` docs with UUID `data.organization.id`), loads each from NATS KV, resolves the canonical **b2b_org Salesforce SFID** from OpenSearch (`object_type=b2b_org`, matched by `data.primary_domain` / `data.website` / `data.name`), and updates through the writer orchestrator (reindexes + fixes the by-organization secondary index).
+
+Use `--committee-uid` or `--member-uid` to scope via indexed NATS reads instead of the OpenSearch discovery query.
+
+Tracked in [LFXV2-2647](https://linuxfoundation.atlassian.net/browse/LFXV2-2647).
+
+**Subcommand flags**
+
+| Flag | Default | Description |
+|---|---|---|
+| `--committee-uid` | `""` | Limit repair to members of a single committee |
+| `--member-uid` | `""` | Limit repair to a single committee member |
+| `--opensearch-url` | `$OPENSEARCH_URL` | Override OpenSearch base URL |
+| `--opensearch-index` | `$OPENSEARCH_INDEX` | Override OpenSearch resources index |
+| `--clear-unresolved` | `false` | When SFID cannot be resolved, clear `organization.id` (keep name/website) |
+| `--sleep` | `0` | Wait between each write (e.g. `200ms`, `1s`) |
+| `--dry-run` | `true` | Log planned repairs without writing (pass `--dry-run=false` to apply) |
+
+**Examples**
+
+Dry-run:
+```sh
+NATS_URL=nats://localhost:4222 \
+OPENSEARCH_URL=http://localhost:9200 \
+  committee-cli sync member-cdp-org-id
+```
+
+Apply repairs:
+```sh
+NATS_URL=nats://localhost:4222 \
+OPENSEARCH_URL=http://localhost:9200 \
+  committee-cli sync member-cdp-org-id --dry-run=false --sleep=200ms
+```
+
 ## Building
 
 ### Local binary