feat(controlplane): add attestation bundle cache with layered resolution

Add a NATS KV / in-memory cache for attestation bundles in the
controlplane, consistent with the platform's caching strategy.

The bundle resolution in addAttestationFromBundle now follows a 3-layer
strategy: cache lookup by digest, DB fallback, and CAS download. This
prepares for eventually dropping the attestation/bundle DB columns.

Closes: #3003

Signed-off-by: Miguel Martinez Trivino <migmartri@gmail.com>
Signed-off-by: Miguel Martinez Trivino <miguel@chainloop.dev>

Author: migmartri

app/controlplane/cmd/wire.go

@@ -141,6 +141,7 @@ var cacheProviderSet = wire.NewSet(
 	newMembershipsCache,
 	newClaimsCache,
 	newPolicyEvalBundleCache,
+	newAttestationBundleCache,
 )
 
 func newClaimsCache(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (cache.Cache[*jwt.MapClaims], error) {
@@ -182,6 +183,23 @@ func newPolicyEvalBundleCache(ctx context.Context, rc *natsconn.ReloadableConnec
 	return cache.New[[]byte](opts...)
 }
 
+func newAttestationBundleCache(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (*biz.AttestationBundleCache, error) {
+	l := log.NewHelper(logger)
+	backend := "memory"
+	opts := []cache.Option{cache.WithTTL(5 * 24 * time.Hour), cache.WithLogger(&kratosLogAdapter{h: l}), cache.WithDescription("Cache for attestation bundles")}
+	if rc != nil {
+		backend = "nats"
+		opts = append(opts, cache.WithNATS(rc.Conn, "chainloop-attestation-bundles"))
+		opts = append(opts, cache.WithReconnect(rc.Subscribe(ctx)))
+	}
+	l.Infow("msg", "cache initialized", "bucket", "chainloop-attestation-bundles", "backend", backend, "ttl", "120h")
+	c, err := cache.New[[]byte](opts...)
+	if err != nil {
+		return nil, err
+	}
+	return &biz.AttestationBundleCache{Cache: c}, nil
+}
+
 // kratosLogAdapter adapts kratos log.Helper (Debugw(...interface{})) to cache.Logger (Debugw(string, ...any)).
 type kratosLogAdapter struct{ h *log.Helper }
 

app/controlplane/cmd/wire_gen.go

@@ -34,6 +34,7 @@ var ProviderSet = wire.NewSet(
 	NewUserUseCase,
 	NewRootAccountUseCase,
 	NewWorkflowRunUseCase,
+	wire.Struct(new(WorkflowRunUseCaseOpts), "*"),
 	NewOrganizationUseCase,
 	NewWorkflowContractUseCase,
 	NewCASCredentialsUseCase,

app/controlplane/pkg/biz/biz.go

@@ -65,10 +65,20 @@ func WireTestData(context.Context, *TestDatabase, *testing.T, log.Logger, creden
 			authzConfig,
 			authzUseCaseConfig,
 			biz.NewIndexConfig,
+			newAttestationBundleCache,
+			newNilCASClient,
 		),
 	)
 }
 
+func newAttestationBundleCache() *biz.AttestationBundleCache {
+	return nil
+}
+
+func newNilCASClient() biz.CASClient {
+	return nil
+}
+
 func authzConfig() *authz.Config {
 	return &authz.Config{RolesMap: authz.RolesMap}
 }

app/controlplane/pkg/biz/testhelpers/wire.go

@@ -1,5 +1,5 @@
 //
-// Copyright 2024-2025 The Chainloop Authors.
+// Copyright 2024-2026 The Chainloop Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -30,6 +30,7 @@ import (
 	"github.com/chainloop-dev/chainloop/pkg/attestation"
 	"github.com/chainloop-dev/chainloop/pkg/attestation/renderer/chainloop"
 	"github.com/chainloop-dev/chainloop/pkg/attestation/verifier"
+	"github.com/chainloop-dev/chainloop/pkg/cache"
 	"github.com/secure-systems-lab/go-securesystemslib/dsse"
 	protobundle "github.com/sigstore/protobuf-specs/gen/pb-go/bundle/v1"
 	"github.com/sigstore/sigstore/pkg/cryptoutils"
@@ -103,24 +104,50 @@ type WorkflowRunRepo interface {
 	Expire(ctx context.Context, id uuid.UUID) error
 }
 
+// AttestationBundleCache wraps cache.Cache[[]byte] to disambiguate from other
+// []byte caches (e.g. policy evaluation bundles) in the wire dependency graph.
+type AttestationBundleCache struct {
+	cache.Cache[[]byte]
+}
+
 type WorkflowRunUseCase struct {
 	wfRunRepo WorkflowRunRepo
 	wfRepo    WorkflowRepo
 	logger    *log.Helper
 	auditorUC *AuditorUseCase
 
 	signingUseCase *SigningUseCase
+	bundleCache    *AttestationBundleCache
+	casClient      CASClient
+	casMappingUC   *CASMappingUseCase
+}
+
+type WorkflowRunUseCaseOpts struct {
+	WfrRepo      WorkflowRunRepo
+	WfRepo       WorkflowRepo
+	SigningUC    *SigningUseCase
+	AuditorUC    *AuditorUseCase
+	Logger       log.Logger
+	BundleCache  *AttestationBundleCache
+	CASClient    CASClient
+	CASMappingUC *CASMappingUseCase
 }
 
-func NewWorkflowRunUseCase(wfrRepo WorkflowRunRepo, wfRepo WorkflowRepo, suc *SigningUseCase, auditorUC *AuditorUseCase, logger log.Logger) (*WorkflowRunUseCase, error) {
+func NewWorkflowRunUseCase(opts *WorkflowRunUseCaseOpts) (*WorkflowRunUseCase, error) {
+	logger := opts.Logger
 	if logger == nil {
 		logger = log.NewStdLogger(io.Discard)
 	}
 
 	return &WorkflowRunUseCase{
-		wfRunRepo: wfrRepo, wfRepo: wfRepo, auditorUC: auditorUC,
-		signingUseCase: suc,
+		wfRunRepo:      opts.WfrRepo,
+		wfRepo:         opts.WfRepo,
+		auditorUC:      opts.AuditorUC,
+		signingUseCase: opts.SigningUC,
 		logger:         log.NewHelper(logger),
+		bundleCache:    opts.BundleCache,
+		casClient:      opts.CASClient,
+		casMappingUC:   opts.CASMappingUC,
 	}, nil
 }
 
@@ -535,29 +562,97 @@ func (uc *WorkflowRunUseCase) GetByDigestInOrgOrPublic(ctx context.Context, orgI
 	return workflowRunInOrgOrPublic(wfrun, orgUUID)
 }
 
+// addAttestationFromBundle resolves the attestation bundle using cache → DB → CAS fallback.
+// Bundles are being migrated from the workflow run DB column into CAS; the layered resolution
+// provides backward compatibility and prepares for dropping the DB column.
 func (uc *WorkflowRunUseCase) addAttestationFromBundle(ctx context.Context, wfRun *WorkflowRun) error {
-	// missing workflow run or attestation already there, do nothing
 	if wfRun == nil || wfRun.State != string(WorkflowRunSuccess) {
 		return nil
 	}
 
-	var bundle protobundle.Bundle
-	bundleBytes, err := uc.wfRunRepo.GetBundle(ctx, wfRun.ID)
-	if err != nil {
-		if IsNotFound(err) {
-			return nil
+	if wfRun.Attestation == nil || wfRun.Attestation.Digest == "" {
+		return nil
+	}
+
+	digest := wfRun.Attestation.Digest
+
+	// Layer 1: cache lookup by digest
+	if uc.bundleCache != nil {
+		cached, found, err := uc.bundleCache.Get(ctx, digest)
+		if err != nil {
+			uc.logger.Warnw("msg", "attestation bundle cache get error", "digest", digest, "error", err)
 		}
+		if found {
+			return uc.applyBundle(wfRun, cached)
+		}
+	}
+
+	// Layer 2: DB fallback
+	bundleBytes, err := uc.wfRunRepo.GetBundle(ctx, wfRun.ID)
+	if err != nil && !IsNotFound(err) {
 		return fmt.Errorf("retrieving bundle from repo: %w", err)
 	}
-	if err = protojson.Unmarshal(bundleBytes, &bundle); err != nil {
+
+	if len(bundleBytes) > 0 {
+		uc.cacheBundle(ctx, digest, bundleBytes)
+		return uc.applyBundle(wfRun, bundleBytes)
+	}
+
+	// Layer 3: CAS download by digest
+	bundleBytes, err = uc.downloadBundleFromCAS(ctx, digest, wfRun.Workflow.OrgID)
+	if err != nil {
+		uc.logger.Warnw("msg", "failed to download bundle from CAS", "digest", digest, "error", err)
+		return nil
+	}
+
+	if len(bundleBytes) > 0 {
+		uc.cacheBundle(ctx, digest, bundleBytes)
+		return uc.applyBundle(wfRun, bundleBytes)
+	}
+
+	return nil
+}
+
+func (uc *WorkflowRunUseCase) applyBundle(wfRun *WorkflowRun, bundleBytes []byte) error {
+	var bundle protobundle.Bundle
+	if err := protojson.Unmarshal(bundleBytes, &bundle); err != nil {
 		return fmt.Errorf("unmarshalling bundle: %w", err)
 	}
 	wfRun.Attestation.Envelope = attestation.DSSEEnvelopeFromBundle(&bundle)
 	wfRun.Attestation.Bundle = bundleBytes
-
 	return nil
 }
 
+func (uc *WorkflowRunUseCase) cacheBundle(ctx context.Context, digest string, data []byte) {
+	if uc.bundleCache == nil {
+		return
+	}
+	if err := uc.bundleCache.Set(ctx, digest, data); err != nil {
+		uc.logger.Warnw("msg", "failed to cache attestation bundle", "digest", digest, "error", err)
+	}
+}
+
+func (uc *WorkflowRunUseCase) downloadBundleFromCAS(ctx context.Context, digest string, orgID uuid.UUID) ([]byte, error) {
+	if uc.casClient == nil || uc.casMappingUC == nil {
+		return nil, nil
+	}
+
+	mapping, err := uc.casMappingUC.FindCASMappingForDownloadByOrg(ctx, digest, []uuid.UUID{orgID}, nil)
+	if err != nil {
+		return nil, fmt.Errorf("finding CAS mapping: %w", err)
+	}
+	if mapping == nil {
+		return nil, nil
+	}
+
+	var buf bytes.Buffer
+	if err := uc.casClient.Download(ctx, string(mapping.CASBackend.Provider), mapping.CASBackend.SecretName, &buf, digest); err != nil {
+		return nil, fmt.Errorf("downloading attestation bundle: %w", err)
+	}
+
+	return buf.Bytes(), nil
+}
+
 // filter the workflow runs that belong to the org or are public
 func workflowRunInOrgOrPublic(wfRun *WorkflowRun, orgID uuid.UUID) (*WorkflowRun, error) {
 	if wfRun == nil || (wfRun.Workflow.OrgID != orgID && !wfRun.Workflow.Public) {

app/controlplane/pkg/biz/testhelpers/wire_gen.go

@@ -1,5 +1,5 @@
 //
-// Copyright 2024 The Chainloop Authors.
+// Copyright 2024-2026 The Chainloop Authors.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -53,7 +53,7 @@ type workflowrunTestSuite struct {
 
 func (s *workflowrunTestSuite) SetupTest() {
 	s.repo = repoM.NewWorkflowRunRepo(s.T())
-	uc, err := biz.NewWorkflowRunUseCase(s.repo, nil, nil, nil, nil)
+	uc, err := biz.NewWorkflowRunUseCase(&biz.WorkflowRunUseCaseOpts{WfrRepo: s.repo})
 	require.NoError(s.T(), err)
 	s.useCase = uc
 	s.validID = uuid.New()