[LFXV2-2190, LFXV2-2193] fix: lazy-bind weekly-brief KV buckets and fix query-service version param

internal/infrastructure/m2m/mailing_list_source.go

@@ -84,6 +84,7 @@ func (m *MailingListSource) ListMailingListActivityForWindow(ctx context.Context
 	}
 	u.Path = appendPath(u.Path, "/query/resources")
 	q := u.Query()
+	q.Set("v", "1")
 	q.Set("type", m.cfg.Type)
 	q.Set("tags", "committee:"+committeeUID)
 	q.Set("start_time[gte]", windowStart.UTC().Format(time.RFC3339Nano))

internal/infrastructure/m2m/meeting_source.go

@@ -97,6 +97,7 @@ func (m *MeetingSource) ListMeetingsForWindow(ctx context.Context, committeeUID
 	}
 	u.Path = appendPath(u.Path, "/query/resources")
 	q := u.Query()
+	q.Set("v", "1")
 	q.Set("type", "v1_past_meeting")
 	q.Set("tags", "committee:"+committeeUID)
 	q.Set("start_time[gte]", windowStart.UTC().Format(time.RFC3339Nano))

internal/infrastructure/m2m/vote_source.go

@@ -84,6 +84,7 @@ func (v *VoteSource) ListVoteActivityForWindow(ctx context.Context, committeeUID
 	}
 	u.Path = appendPath(u.Path, "/query/resources")
 	q := u.Query()
+	q.Set("v", "1")
 	q.Set("type", v.cfg.Type)
 	q.Set("tags", "committee:"+committeeUID)
 	q.Set("start_time[gte]", windowStart.UTC().Format(time.RFC3339Nano))

internal/infrastructure/nats/client.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"sync"
 	"time"
 
 	"github.com/linuxfoundation/lfx-v2-committee-service/internal/domain/port"
@@ -23,11 +24,13 @@ import (
 
 // NATSClient wraps the NATS connection and provides access control operations
 type NATSClient struct {
-	conn     *nats.Conn
-	config   Config
-	kvStore  map[string]jetstream.KeyValue
-	objStore map[string]jetstream.ObjectStore
-	timeout  time.Duration
+	conn        *nats.Conn
+	config      Config
+	kvMu        sync.RWMutex
+	kvStore     map[string]jetstream.KeyValue
+	kvStoreLazy map[string]jetstream.KeyValue
+	objStore    map[string]jetstream.ObjectStore
+	timeout     time.Duration
 }
 
 // NATSClientInterface defines the interface for NATS operations
@@ -83,6 +86,47 @@ func (c *NATSClient) KeyValueStore(ctx context.Context, bucketName string) error
 	return nil
 }
 
+// GetOrBindKVStore returns the KV handle for bucketName. If the bucket was
+// absent at startup (e.g. a pod that booted before the bucket was provisioned),
+// it attempts to bind on first access so the pod self-heals without a restart.
+//
+// Lazy handles are kept in kvStoreLazy, separate from the startup-populated
+// kvStore. kvStore is written only during NewClient (single-threaded) so the
+// 50+ call sites that read it directly remain race-free. kvStoreLazy is always
+// accessed under kvMu. The NATS bind call itself runs outside the lock so
+// concurrent cache-hit lookups are never blocked by an in-flight bind.
+func (c *NATSClient) GetOrBindKVStore(ctx context.Context, bucketName string) (jetstream.KeyValue, error) {
+	c.kvMu.RLock()
+	kv, ok := c.kvStoreLazy[bucketName]
+	c.kvMu.RUnlock()
+	if ok {
+		return kv, nil
+	}
+
+	// Bind outside the lock — js.KeyValue is a network call and must not hold
+	// kvMu while it waits, which would block concurrent cache-hit lookups.
+	js, err := jetstream.New(c.conn)
+	if err != nil {
+		return nil, errors.NewServiceUnavailable("failed to create JetStream client", err)
+	}
+	kv, err = js.KeyValue(ctx, bucketName)
+	if err != nil {
+		return nil, errors.NewServiceUnavailable(fmt.Sprintf("%s bucket not initialized", bucketName), err)
+	}
+
+	c.kvMu.Lock()
+	defer c.kvMu.Unlock()
+	if existing, ok := c.kvStoreLazy[bucketName]; ok {
+		return existing, nil
+	}
+	if c.kvStoreLazy == nil {
+		c.kvStoreLazy = make(map[string]jetstream.KeyValue)
+	}
+	c.kvStoreLazy[bucketName] = kv
+	slog.InfoContext(ctx, "weekly-brief KV bucket bound on first access", "bucket", bucketName)
+	return kv, nil
+}
+
 // ObjectStore creates a JetStream client and gets the object store by name.
 func (c *NATSClient) ObjectStore(ctx context.Context, storeName string) error {
 	js, err := jetstream.New(c.conn)
@@ -272,27 +316,6 @@ func NewClient(ctx context.Context, config Config) (*NATSClient, error) {
 		)
 	}
 
-	// Weekly-brief buckets are initialized best-effort. If they aren't yet
-	// provisioned (e.g. a rolling deploy where the chart hasn't created them, or
-	// a local NATS without them) the service still starts; only the weekly-brief
-	// endpoints return ServiceUnavailable until the buckets exist.
-	for _, bucketName := range []string{
-		constants.KVBucketNameGroupWeeklyBriefs,
-		constants.KVBucketNameGroupWeeklyBriefUIDIndex,
-		constants.KVBucketNameGroupWeeklyBriefThrottle,
-	} {
-		if err := client.KeyValueStore(ctx, bucketName); err != nil {
-			slog.WarnContext(ctx, "weekly-brief KV bucket not initialized; weekly-brief endpoints will be unavailable until it is provisioned",
-				"error", err,
-				"bucket", bucketName,
-			)
-			continue
-		}
-		slog.InfoContext(ctx, "NATS key-value store initialized",
-			"bucket", bucketName,
-		)
-	}
-
 	for _, storeName := range []string{
 		constants.ObjectStoreNameCommitteeDocuments,
 	} {

internal/infrastructure/nats/group_weekly_brief_storage.go

@@ -47,9 +47,9 @@ func (s *storage) GetGroupWeeklyBriefForWindow(ctx context.Context, committeeUID
 	// WindowStart; the rest of the fields are unused.
 	indexKey := buildBriefIndexKey(committeeUID, model.WindowDateKey(windowStart.WindowStart))
 
-	idxBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefUIDIndex]
-	if !ok {
-		return nil, nil, errs.NewServiceUnavailable("group-weekly-brief-uid-index bucket not initialized")
+	idxBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefUIDIndex)
+	if err != nil {
+		return nil, nil, err
 	}
 
 	entry, err := idxBucket.Get(ctx, indexKey)
@@ -64,9 +64,9 @@ func (s *storage) GetGroupWeeklyBriefForWindow(ctx context.Context, committeeUID
 		return nil, nil, nil
 	}
 
-	briefBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefs]
-	if !ok {
-		return nil, nil, errs.NewServiceUnavailable("group-weekly-briefs bucket not initialized")
+	briefBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefs)
+	if err != nil {
+		return nil, nil, err
 	}
 	briefEntry, errGet := briefBucket.Get(ctx, sanitizeKVKey(briefUID))
 	if errGet != nil {
@@ -107,7 +107,7 @@ func (s *storage) GetGroupWeeklyBriefForWindow(ctx context.Context, committeeUID
 	// Best-effort throttle lookup. Misses and errors don't fail the read —
 	// throttle is advisory metadata.
 	var throttleBytes []byte
-	if thBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefThrottle]; ok {
+	if thBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefThrottle); err == nil {
 		thEntry, thErr := thBucket.Get(ctx, indexKey)
 		switch {
 		case thErr == nil:
@@ -153,13 +153,13 @@ func (s *storage) PutGroupWeeklyBrief(ctx context.Context, brief *model.GroupWee
 	}
 	brief.UpdatedAt = now
 
-	briefBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefs]
-	if !ok {
-		return nil, errs.NewServiceUnavailable("group-weekly-briefs bucket not initialized")
+	briefBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefs)
+	if err != nil {
+		return nil, err
 	}
-	idxBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefUIDIndex]
-	if !ok {
-		return nil, errs.NewServiceUnavailable("group-weekly-brief-uid-index bucket not initialized")
+	idxBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefUIDIndex)
+	if err != nil {
+		return nil, err
 	}
 
 	payload, err := json.Marshal(brief)
@@ -211,9 +211,9 @@ func (s *storage) PutGroupWeeklyBrief(ctx context.Context, brief *model.GroupWee
 // GetGroupWeeklyBriefThrottle returns the throttle entry for the given
 // (committee, window-start) pair. A miss returns (nil, nil).
 func (s *storage) GetGroupWeeklyBriefThrottle(ctx context.Context, committeeUID string, windowStart time.Time) (*model.GroupWeeklyBriefThrottle, error) {
-	thBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefThrottle]
-	if !ok {
-		return nil, errs.NewServiceUnavailable("group-weekly-brief-throttle bucket not initialized")
+	thBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefThrottle)
+	if err != nil {
+		return nil, err
 	}
 	key := buildBriefIndexKey(committeeUID, model.WindowDateKey(windowStart))
 	entry, err := thBucket.Get(ctx, key)
@@ -245,9 +245,9 @@ func (s *storage) PutGroupWeeklyBriefThrottle(ctx context.Context, throttle *mod
 	if throttle.WindowStart.IsZero() {
 		return nil, errs.NewValidation("window_start is required")
 	}
-	thBucket, ok := s.client.kvStore[constants.KVBucketNameGroupWeeklyBriefThrottle]
-	if !ok {
-		return nil, errs.NewServiceUnavailable("group-weekly-brief-throttle bucket not initialized")
+	thBucket, err := s.client.GetOrBindKVStore(ctx, constants.KVBucketNameGroupWeeklyBriefThrottle)
+	if err != nil {
+		return nil, err
 	}
 
 	payload, err := json.Marshal(throttle)