From a699e8eb81b6ea797d578bacf7e332a902bc7b49 Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Wed, 8 Apr 2026 14:40:53 +0200 Subject: [PATCH 1/2] 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: chainloop-dev/chainloop#3003 Signed-off-by: Miguel Martinez Trivino Signed-off-by: Miguel Martinez Trivino --- app/controlplane/cmd/wire.go | 18 +++ app/controlplane/cmd/wire_gen.go | 36 +++++- app/controlplane/pkg/biz/biz.go | 1 + app/controlplane/pkg/biz/testhelpers/wire.go | 10 ++ .../pkg/biz/testhelpers/wire_gen.go | 26 +++- app/controlplane/pkg/biz/workflowrun.go | 119 ++++++++++++++++-- app/controlplane/pkg/biz/workflowrun_test.go | 4 +- 7 files changed, 196 insertions(+), 18 deletions(-) diff --git a/app/controlplane/cmd/wire.go b/app/controlplane/cmd/wire.go index cada02b2b..c518b9ddd 100644 --- a/app/controlplane/cmd/wire.go +++ b/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 } diff --git a/app/controlplane/cmd/wire_gen.go b/app/controlplane/cmd/wire_gen.go index 271b72ce9..29b80ce2d 100644 --- a/app/controlplane/cmd/wire_gen.go +++ b/app/controlplane/cmd/wire_gen.go @@ -188,7 +188,7 @@ func wireApp(contextContext context.Context, bootstrap *conf.Bootstrap, readerWr cleanup() return nil, nil, err } - workflowRunUseCase, err := biz.NewWorkflowRunUseCase(workflowRunRepo, workflowRepo, signingUseCase, auditorUseCase, logger) + attestationBundleCache, err := newAttestationBundleCache(contextContext, reloadableConnection, logger) if err != nil { cleanup2() cleanup() @@ -196,6 +196,22 @@ func wireApp(contextContext context.Context, bootstrap *conf.Bootstrap, readerWr } casMappingRepo := data.NewCASMappingRepo(dataData, casBackendRepo, logger) casMappingUseCase := biz.NewCASMappingUseCase(casMappingRepo, membershipUseCase, logger) + workflowRunUseCaseOpts := &biz.WorkflowRunUseCaseOpts{ + WfrRepo: workflowRunRepo, + WfRepo: workflowRepo, + SigningUC: signingUseCase, + AuditorUC: auditorUseCase, + Logger: logger, + BundleCache: attestationBundleCache, + CASClient: casClientUseCase, + CASMappingUC: casMappingUseCase, + } + workflowRunUseCase, err := biz.NewWorkflowRunUseCase(workflowRunUseCaseOpts) + if err != nil { + cleanup2() + cleanup() + return nil, nil, err + } cache2, err := newPolicyEvalBundleCache(contextContext, reloadableConnection, logger) if err != nil { cleanup2() @@ -432,6 +448,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) { @@ -473,6 +490,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 } diff --git a/app/controlplane/pkg/biz/biz.go b/app/controlplane/pkg/biz/biz.go index 1951c285a..497db0400 100644 --- a/app/controlplane/pkg/biz/biz.go +++ b/app/controlplane/pkg/biz/biz.go @@ -34,6 +34,7 @@ var ProviderSet = wire.NewSet( NewUserUseCase, NewRootAccountUseCase, NewWorkflowRunUseCase, + wire.Struct(new(WorkflowRunUseCaseOpts), "*"), NewOrganizationUseCase, NewWorkflowContractUseCase, NewCASCredentialsUseCase, diff --git a/app/controlplane/pkg/biz/testhelpers/wire.go b/app/controlplane/pkg/biz/testhelpers/wire.go index 9f8168620..b78a4b10e 100644 --- a/app/controlplane/pkg/biz/testhelpers/wire.go +++ b/app/controlplane/pkg/biz/testhelpers/wire.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} } diff --git a/app/controlplane/pkg/biz/testhelpers/wire_gen.go b/app/controlplane/pkg/biz/testhelpers/wire_gen.go index 09e4546d6..c82f863be 100644 --- a/app/controlplane/pkg/biz/testhelpers/wire_gen.go +++ b/app/controlplane/pkg/biz/testhelpers/wire_gen.go @@ -95,7 +95,21 @@ func WireTestData(contextContext context.Context, testDatabase *TestDatabase, t cleanup() return nil, nil, err } - workflowRunUseCase, err := biz.NewWorkflowRunUseCase(workflowRunRepo, workflowRepo, signingUseCase, auditorUseCase, logger) + attestationBundleCache := newAttestationBundleCache() + casClient := newNilCASClient() + casMappingRepo := data.NewCASMappingRepo(dataData, casBackendRepo, logger) + casMappingUseCase := biz.NewCASMappingUseCase(casMappingRepo, membershipUseCase, logger) + workflowRunUseCaseOpts := &biz.WorkflowRunUseCaseOpts{ + WfrRepo: workflowRunRepo, + WfRepo: workflowRepo, + SigningUC: signingUseCase, + AuditorUC: auditorUseCase, + Logger: logger, + BundleCache: attestationBundleCache, + CASClient: casClient, + CASMappingUC: casMappingUseCase, + } + workflowRunUseCase, err := biz.NewWorkflowRunUseCase(workflowRunUseCaseOpts) if err != nil { cleanup() return nil, nil, err @@ -122,8 +136,6 @@ func WireTestData(contextContext context.Context, testDatabase *TestDatabase, t userUseCase := biz.NewUserUseCase(newUserUseCaseParams) robotAccountRepo := data.NewRobotAccountRepo(dataData, logger) robotAccountUseCase := biz.NewRootAccountUseCase(robotAccountRepo, workflowRepo, auth, logger) - casMappingRepo := data.NewCASMappingRepo(dataData, casBackendRepo, logger) - casMappingUseCase := biz.NewCASMappingUseCase(casMappingRepo, membershipUseCase, logger) orgInvitationRepo := data.NewOrgInvitation(dataData, logger) orgInvitationUseCase, err := biz.NewOrgInvitationUseCase(orgInvitationRepo, membershipRepo, userRepo, auditorUseCase, groupRepo, projectsRepo, logger) if err != nil { @@ -207,6 +219,14 @@ var ( // wire.go: +func newAttestationBundleCache() *biz.AttestationBundleCache { + return nil +} + +func newNilCASClient() biz.CASClient { + return nil +} + func authzConfig() *authz.Config { return &authz.Config{RolesMap: authz.RolesMap} } diff --git a/app/controlplane/pkg/biz/workflowrun.go b/app/controlplane/pkg/biz/workflowrun.go index 02e6e29ac..fb07fc3cf 100644 --- a/app/controlplane/pkg/biz/workflowrun.go +++ b/app/controlplane/pkg/biz/workflowrun.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,6 +104,12 @@ 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 @@ -110,17 +117,37 @@ type WorkflowRunUseCase struct { 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) { diff --git a/app/controlplane/pkg/biz/workflowrun_test.go b/app/controlplane/pkg/biz/workflowrun_test.go index 3c9894a7c..6906d7495 100644 --- a/app/controlplane/pkg/biz/workflowrun_test.go +++ b/app/controlplane/pkg/biz/workflowrun_test.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() From 403661fe891414607763f32b2510cc76d0f0b9c0 Mon Sep 17 00:00:00 2001 From: Miguel Martinez Trivino Date: Wed, 8 Apr 2026 15:24:59 +0200 Subject: [PATCH 2/2] refactor: extract cache wrapper types into pkg/cache sub-packages Move AttestationBundleCache and policy eval bundle cache types into their own packages (pkg/cache/attestationbundle and pkg/cache/policyevalbundle) so they can be initialized outside of the controlplane. Each package provides a Cache wrapper type, a New constructor, and default constants for TTL, bucket name, and description. Signed-off-by: Miguel Martinez Trivino Signed-off-by: Miguel Martinez Trivino --- app/controlplane/cmd/wire.go | 56 ++-------------- app/controlplane/cmd/wire_gen.go | 67 +++---------------- .../internal/service/workflowrun.go | 6 +- app/controlplane/pkg/biz/testhelpers/wire.go | 3 +- .../pkg/biz/testhelpers/wire_gen.go | 7 +- app/controlplane/pkg/biz/workflowrun.go | 12 +--- .../attestationbundle/attestationbundle.go | 60 +++++++++++++++++ pkg/cache/cache.go | 19 +++--- pkg/cache/memory.go | 10 +-- pkg/cache/natskv.go | 32 ++++----- .../policyevalbundle/policyevalbundle.go | 60 +++++++++++++++++ 11 files changed, 177 insertions(+), 155 deletions(-) create mode 100644 pkg/cache/attestationbundle/attestationbundle.go create mode 100644 pkg/cache/policyevalbundle/policyevalbundle.go diff --git a/app/controlplane/cmd/wire.go b/app/controlplane/cmd/wire.go index c518b9ddd..8f42f98db 100644 --- a/app/controlplane/cmd/wire.go +++ b/app/controlplane/cmd/wire.go @@ -38,6 +38,8 @@ import ( "github.com/chainloop-dev/chainloop/app/controlplane/plugins/sdk/v1" "github.com/chainloop-dev/chainloop/pkg/blobmanager/loader" "github.com/chainloop-dev/chainloop/pkg/cache" + "github.com/chainloop-dev/chainloop/pkg/cache/attestationbundle" + "github.com/chainloop-dev/chainloop/pkg/cache/policyevalbundle" "github.com/chainloop-dev/chainloop/pkg/credentials" "github.com/chainloop-dev/chainloop/pkg/natsconn" "github.com/go-kratos/kratos/v2/log" @@ -140,14 +142,14 @@ func newAuthAllowList(conf *conf.Bootstrap) *pkgConf.AllowList { var cacheProviderSet = wire.NewSet( newMembershipsCache, newClaimsCache, - newPolicyEvalBundleCache, - newAttestationBundleCache, + policyevalbundle.New, + attestationbundle.New, ) func newClaimsCache(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (cache.Cache[*jwt.MapClaims], error) { l := log.NewHelper(logger) backend := "memory" - opts := []cache.Option{cache.WithTTL(10 * time.Second), cache.WithLogger(&kratosLogAdapter{h: l}), cache.WithDescription("Cache for JWT claims")} + opts := []cache.Option{cache.WithTTL(10 * time.Second), cache.WithLogger(l), cache.WithDescription("Cache for JWT claims")} if rc != nil { backend = "nats" opts = append(opts, cache.WithNATS(rc.Conn, "chainloop-jwt-claims")) @@ -160,7 +162,7 @@ func newClaimsCache(ctx context.Context, rc *natsconn.ReloadableConnection, logg func newMembershipsCache(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (cache.Cache[*entities.Membership], error) { l := log.NewHelper(logger) backend := "memory" - opts := []cache.Option{cache.WithTTL(time.Second), cache.WithLogger(&kratosLogAdapter{h: l}), cache.WithDescription("Cache for org memberships")} + opts := []cache.Option{cache.WithTTL(time.Second), cache.WithLogger(l), cache.WithDescription("Cache for org memberships")} if rc != nil { backend = "nats" opts = append(opts, cache.WithNATS(rc.Conn, "chainloop-memberships")) @@ -169,49 +171,3 @@ func newMembershipsCache(ctx context.Context, rc *natsconn.ReloadableConnection, l.Infow("msg", "cache initialized", "bucket", "chainloop-memberships", "backend", backend, "ttl", "1s") return cache.New[*entities.Membership](opts...) } - -func newPolicyEvalBundleCache(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (cache.Cache[[]byte], error) { - l := log.NewHelper(logger) - backend := "memory" - opts := []cache.Option{cache.WithTTL(24 * time.Hour), cache.WithLogger(&kratosLogAdapter{h: l}), cache.WithDescription("Cache for policy evaluation bundles from CAS")} - if rc != nil { - backend = "nats" - opts = append(opts, cache.WithNATS(rc.Conn, "chainloop-policy-eval-bundles")) - opts = append(opts, cache.WithReconnect(rc.Subscribe(ctx))) - } - l.Infow("msg", "cache initialized", "bucket", "chainloop-policy-eval-bundles", "backend", backend, "ttl", "24h") - 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 } - -func (a *kratosLogAdapter) Debugw(msg string, keyvals ...any) { - a.h.Debugw(append([]any{"msg", msg}, keyvals...)...) -} -func (a *kratosLogAdapter) Infow(msg string, keyvals ...any) { - a.h.Infow(append([]any{"msg", msg}, keyvals...)...) -} -func (a *kratosLogAdapter) Warnw(msg string, keyvals ...any) { - a.h.Warnw(append([]any{"msg", msg}, keyvals...)...) -} -func (a *kratosLogAdapter) Errorw(msg string, keyvals ...any) { - a.h.Errorw(append([]any{"msg", msg}, keyvals...)...) -} diff --git a/app/controlplane/cmd/wire_gen.go b/app/controlplane/cmd/wire_gen.go index 29b80ce2d..3e3f85384 100644 --- a/app/controlplane/cmd/wire_gen.go +++ b/app/controlplane/cmd/wire_gen.go @@ -22,6 +22,8 @@ import ( "github.com/chainloop-dev/chainloop/app/controlplane/plugins/sdk/v1" "github.com/chainloop-dev/chainloop/pkg/blobmanager/loader" "github.com/chainloop-dev/chainloop/pkg/cache" + "github.com/chainloop-dev/chainloop/pkg/cache/attestationbundle" + "github.com/chainloop-dev/chainloop/pkg/cache/policyevalbundle" "github.com/chainloop-dev/chainloop/pkg/credentials" "github.com/chainloop-dev/chainloop/pkg/natsconn" "github.com/go-kratos/kratos/v2/log" @@ -188,7 +190,7 @@ func wireApp(contextContext context.Context, bootstrap *conf.Bootstrap, readerWr cleanup() return nil, nil, err } - attestationBundleCache, err := newAttestationBundleCache(contextContext, reloadableConnection, logger) + attestationbundleCache, err := attestationbundle.New(contextContext, reloadableConnection, logger) if err != nil { cleanup2() cleanup() @@ -202,7 +204,7 @@ func wireApp(contextContext context.Context, bootstrap *conf.Bootstrap, readerWr SigningUC: signingUseCase, AuditorUC: auditorUseCase, Logger: logger, - BundleCache: attestationBundleCache, + BundleCache: attestationbundleCache, CASClient: casClientUseCase, CASMappingUC: casMappingUseCase, } @@ -212,7 +214,7 @@ func wireApp(contextContext context.Context, bootstrap *conf.Bootstrap, readerWr cleanup() return nil, nil, err } - cache2, err := newPolicyEvalBundleCache(contextContext, reloadableConnection, logger) + policyevalbundleCache, err := policyevalbundle.New(contextContext, reloadableConnection, logger) if err != nil { cleanup2() cleanup() @@ -226,7 +228,7 @@ func wireApp(contextContext context.Context, bootstrap *conf.Bootstrap, readerWr CredsReader: readerWriter, CASClient: casClientUseCase, CASMappingUC: casMappingUseCase, - PolicyEvalCache: cache2, + PolicyEvalCache: policyevalbundleCache, Opts: v5, } workflowRunService := service.NewWorkflowRunService(newWorkflowRunServiceOpts) @@ -446,15 +448,13 @@ func newAuthAllowList(conf2 *conf.Bootstrap) *v1.AllowList { var cacheProviderSet = wire.NewSet( newMembershipsCache, - newClaimsCache, - newPolicyEvalBundleCache, - newAttestationBundleCache, + newClaimsCache, policyevalbundle.New, attestationbundle.New, ) func newClaimsCache(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (cache.Cache[*jwt.MapClaims], error) { l := log.NewHelper(logger) backend := "memory" - opts := []cache.Option{cache.WithTTL(10 * time.Second), cache.WithLogger(&kratosLogAdapter{h: l}), cache.WithDescription("Cache for JWT claims")} + opts := []cache.Option{cache.WithTTL(10 * time.Second), cache.WithLogger(l), cache.WithDescription("Cache for JWT claims")} if rc != nil { backend = "nats" opts = append(opts, cache.WithNATS(rc.Conn, "chainloop-jwt-claims")) @@ -467,7 +467,7 @@ func newClaimsCache(ctx context.Context, rc *natsconn.ReloadableConnection, logg func newMembershipsCache(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (cache.Cache[*entities.Membership], error) { l := log.NewHelper(logger) backend := "memory" - opts := []cache.Option{cache.WithTTL(time.Second), cache.WithLogger(&kratosLogAdapter{h: l}), cache.WithDescription("Cache for org memberships")} + opts := []cache.Option{cache.WithTTL(time.Second), cache.WithLogger(l), cache.WithDescription("Cache for org memberships")} if rc != nil { backend = "nats" opts = append(opts, cache.WithNATS(rc.Conn, "chainloop-memberships")) @@ -476,52 +476,3 @@ func newMembershipsCache(ctx context.Context, rc *natsconn.ReloadableConnection, l.Infow("msg", "cache initialized", "bucket", "chainloop-memberships", "backend", backend, "ttl", "1s") return cache.New[*entities.Membership](opts...) } - -func newPolicyEvalBundleCache(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (cache.Cache[[]byte], error) { - l := log.NewHelper(logger) - backend := "memory" - opts := []cache.Option{cache.WithTTL(24 * time.Hour), cache.WithLogger(&kratosLogAdapter{h: l}), cache.WithDescription("Cache for policy evaluation bundles from CAS")} - if rc != nil { - backend = "nats" - opts = append(opts, cache.WithNATS(rc.Conn, "chainloop-policy-eval-bundles")) - opts = append(opts, cache.WithReconnect(rc.Subscribe(ctx))) - } - l.Infow("msg", "cache initialized", "bucket", "chainloop-policy-eval-bundles", "backend", backend, "ttl", "24h") - 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 } - -func (a *kratosLogAdapter) Debugw(msg string, keyvals ...any) { - a.h.Debugw(append([]any{"msg", msg}, keyvals...)...) -} - -func (a *kratosLogAdapter) Infow(msg string, keyvals ...any) { - a.h.Infow(append([]any{"msg", msg}, keyvals...)...) -} - -func (a *kratosLogAdapter) Warnw(msg string, keyvals ...any) { - a.h.Warnw(append([]any{"msg", msg}, keyvals...)...) -} - -func (a *kratosLogAdapter) Errorw(msg string, keyvals ...any) { - a.h.Errorw(append([]any{"msg", msg}, keyvals...)...) -} diff --git a/app/controlplane/internal/service/workflowrun.go b/app/controlplane/internal/service/workflowrun.go index 3f5b39317..66dd3a41d 100644 --- a/app/controlplane/internal/service/workflowrun.go +++ b/app/controlplane/internal/service/workflowrun.go @@ -27,7 +27,7 @@ import ( "github.com/chainloop-dev/chainloop/app/controlplane/pkg/biz" "github.com/chainloop-dev/chainloop/app/controlplane/pkg/pagination" chainloop "github.com/chainloop-dev/chainloop/pkg/attestation/renderer/chainloop" - "github.com/chainloop-dev/chainloop/pkg/cache" + "github.com/chainloop-dev/chainloop/pkg/cache/policyevalbundle" "github.com/chainloop-dev/chainloop/pkg/credentials" errors "github.com/go-kratos/kratos/v2/errors" "github.com/google/uuid" @@ -46,7 +46,7 @@ type WorkflowRunService struct { credsReader credentials.Reader casClient biz.CASClient casMappingUC *biz.CASMappingUseCase - policyEvalCache cache.Cache[[]byte] + policyEvalCache *policyevalbundle.Cache } type NewWorkflowRunServiceOpts struct { @@ -57,7 +57,7 @@ type NewWorkflowRunServiceOpts struct { CredsReader credentials.Reader CASClient biz.CASClient CASMappingUC *biz.CASMappingUseCase - PolicyEvalCache cache.Cache[[]byte] + PolicyEvalCache *policyevalbundle.Cache Opts []NewOpt } diff --git a/app/controlplane/pkg/biz/testhelpers/wire.go b/app/controlplane/pkg/biz/testhelpers/wire.go index b78a4b10e..7c93e1d2c 100644 --- a/app/controlplane/pkg/biz/testhelpers/wire.go +++ b/app/controlplane/pkg/biz/testhelpers/wire.go @@ -35,6 +35,7 @@ import ( "github.com/chainloop-dev/chainloop/app/controlplane/plugins/sdk/v1" robotaccount "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" backends "github.com/chainloop-dev/chainloop/pkg/blobmanager" + "github.com/chainloop-dev/chainloop/pkg/cache/attestationbundle" "github.com/chainloop-dev/chainloop/pkg/credentials" "github.com/chainloop-dev/chainloop/pkg/natsconn" "github.com/go-kratos/kratos/v2/log" @@ -71,7 +72,7 @@ func WireTestData(context.Context, *TestDatabase, *testing.T, log.Logger, creden ) } -func newAttestationBundleCache() *biz.AttestationBundleCache { +func newAttestationBundleCache() *attestationbundle.Cache { return nil } diff --git a/app/controlplane/pkg/biz/testhelpers/wire_gen.go b/app/controlplane/pkg/biz/testhelpers/wire_gen.go index c82f863be..6799a3dbe 100644 --- a/app/controlplane/pkg/biz/testhelpers/wire_gen.go +++ b/app/controlplane/pkg/biz/testhelpers/wire_gen.go @@ -18,6 +18,7 @@ import ( "github.com/chainloop-dev/chainloop/app/controlplane/plugins/sdk/v1" "github.com/chainloop-dev/chainloop/internal/robotaccount/cas" "github.com/chainloop-dev/chainloop/pkg/blobmanager" + "github.com/chainloop-dev/chainloop/pkg/cache/attestationbundle" "github.com/chainloop-dev/chainloop/pkg/credentials" "github.com/chainloop-dev/chainloop/pkg/natsconn" "github.com/go-kratos/kratos/v2/log" @@ -95,7 +96,7 @@ func WireTestData(contextContext context.Context, testDatabase *TestDatabase, t cleanup() return nil, nil, err } - attestationBundleCache := newAttestationBundleCache() + cache := newAttestationBundleCache() casClient := newNilCASClient() casMappingRepo := data.NewCASMappingRepo(dataData, casBackendRepo, logger) casMappingUseCase := biz.NewCASMappingUseCase(casMappingRepo, membershipUseCase, logger) @@ -105,7 +106,7 @@ func WireTestData(contextContext context.Context, testDatabase *TestDatabase, t SigningUC: signingUseCase, AuditorUC: auditorUseCase, Logger: logger, - BundleCache: attestationBundleCache, + BundleCache: cache, CASClient: casClient, CASMappingUC: casMappingUseCase, } @@ -219,7 +220,7 @@ var ( // wire.go: -func newAttestationBundleCache() *biz.AttestationBundleCache { +func newAttestationBundleCache() *attestationbundle.Cache { return nil } diff --git a/app/controlplane/pkg/biz/workflowrun.go b/app/controlplane/pkg/biz/workflowrun.go index fb07fc3cf..20e0a4b98 100644 --- a/app/controlplane/pkg/biz/workflowrun.go +++ b/app/controlplane/pkg/biz/workflowrun.go @@ -30,7 +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/chainloop-dev/chainloop/pkg/cache/attestationbundle" "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" @@ -104,12 +104,6 @@ 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 @@ -117,7 +111,7 @@ type WorkflowRunUseCase struct { auditorUC *AuditorUseCase signingUseCase *SigningUseCase - bundleCache *AttestationBundleCache + bundleCache *attestationbundle.Cache casClient CASClient casMappingUC *CASMappingUseCase } @@ -128,7 +122,7 @@ type WorkflowRunUseCaseOpts struct { SigningUC *SigningUseCase AuditorUC *AuditorUseCase Logger log.Logger - BundleCache *AttestationBundleCache + BundleCache *attestationbundle.Cache CASClient CASClient CASMappingUC *CASMappingUseCase } diff --git a/pkg/cache/attestationbundle/attestationbundle.go b/pkg/cache/attestationbundle/attestationbundle.go new file mode 100644 index 000000000..fca38eda6 --- /dev/null +++ b/pkg/cache/attestationbundle/attestationbundle.go @@ -0,0 +1,60 @@ +// +// Copyright 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. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package attestationbundle provides a typed cache for attestation bundles. +package attestationbundle + +import ( + "context" + "time" + + "github.com/chainloop-dev/chainloop/pkg/cache" + "github.com/chainloop-dev/chainloop/pkg/natsconn" + "github.com/go-kratos/kratos/v2/log" +) + +const ( + ttl = 5 * 24 * time.Hour + bucket = "chainloop-attestation-bundles" + description = "Cache for attestation bundles" +) + +// Cache wraps cache.Cache[[]byte] to provide a distinct type for wire disambiguation. +type Cache struct { + cache.Cache[[]byte] +} + +// New creates an attestation bundle cache with built-in TTL, bucket, and description. +func New(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (*Cache, error) { + opts := []cache.Option{ + cache.WithTTL(ttl), + cache.WithDescription(description), + } + + if logger != nil { + opts = append(opts, cache.WithLogger(log.NewHelper(logger))) + } + + if rc != nil { + opts = append(opts, cache.WithNATS(rc.Conn, bucket)) + opts = append(opts, cache.WithReconnect(rc.Subscribe(ctx))) + } + + c, err := cache.New[[]byte](opts...) + if err != nil { + return nil, err + } + return &Cache{Cache: c}, nil +} diff --git a/pkg/cache/cache.go b/pkg/cache/cache.go index 8fdb4cb76..1d69224e6 100644 --- a/pkg/cache/cache.go +++ b/pkg/cache/cache.go @@ -32,13 +32,12 @@ type Cache[T any] interface { Purge(ctx context.Context) error } -// Logger is a structured logging interface compatible with Kratos log.Helper -// but does not import it. +// Logger is a structured logging interface satisfied by *log.Helper from Kratos. type Logger interface { - Debugw(msg string, keyvals ...any) - Warnw(msg string, keyvals ...any) - Infow(msg string, keyvals ...any) - Errorw(msg string, keyvals ...any) + Debugw(keyvals ...any) + Warnw(keyvals ...any) + Infow(keyvals ...any) + Errorw(keyvals ...any) } // defaultMaxSize is a sensible upper bound on in-memory cache entries @@ -119,7 +118,7 @@ func New[T any](opts ...Option) (Cache[T], error) { // nopLogger is a no-op implementation of Logger. type nopLogger struct{} -func (nopLogger) Debugw(_ string, _ ...any) {} -func (nopLogger) Warnw(_ string, _ ...any) {} -func (nopLogger) Infow(_ string, _ ...any) {} -func (nopLogger) Errorw(_ string, _ ...any) {} +func (nopLogger) Debugw(_ ...any) {} +func (nopLogger) Warnw(_ ...any) {} +func (nopLogger) Infow(_ ...any) {} +func (nopLogger) Errorw(_ ...any) {} diff --git a/pkg/cache/memory.go b/pkg/cache/memory.go index 775b7943b..8e444f6ff 100644 --- a/pkg/cache/memory.go +++ b/pkg/cache/memory.go @@ -27,7 +27,7 @@ type memoryCache[T any] struct { } func newMemoryCache[T any](cfg *config) *memoryCache[T] { - cfg.logger.Infow("cache: using in-memory LRU backend", "ttl", cfg.ttl, "maxSize", cfg.maxSize) + cfg.logger.Infow("msg", "cache: using in-memory LRU backend", "ttl", cfg.ttl, "maxSize", cfg.maxSize) return &memoryCache[T]{ lru: expirable.NewLRU[string, T](cfg.maxSize, nil, cfg.ttl), logger: cfg.logger, @@ -36,24 +36,24 @@ func newMemoryCache[T any](cfg *config) *memoryCache[T] { func (m *memoryCache[T]) Get(_ context.Context, key string) (T, bool, error) { val, ok := m.lru.Get(key) - m.logger.Debugw("cache get", "key", key, "hit", ok, "backend", "memory") + m.logger.Debugw("msg", "cache get", "key", key, "hit", ok, "backend", "memory") return val, ok, nil } func (m *memoryCache[T]) Set(_ context.Context, key string, value T) error { m.lru.Add(key, value) - m.logger.Debugw("cache set", "key", key, "backend", "memory") + m.logger.Debugw("msg", "cache set", "key", key, "backend", "memory") return nil } func (m *memoryCache[T]) Delete(_ context.Context, key string) error { m.lru.Remove(key) - m.logger.Debugw("cache delete", "key", key, "backend", "memory") + m.logger.Debugw("msg", "cache delete", "key", key, "backend", "memory") return nil } func (m *memoryCache[T]) Purge(_ context.Context) error { m.lru.Purge() - m.logger.Debugw("cache purge", "backend", "memory") + m.logger.Debugw("msg", "cache purge", "backend", "memory") return nil } diff --git a/pkg/cache/natskv.go b/pkg/cache/natskv.go index f9e275142..72e53a5dd 100644 --- a/pkg/cache/natskv.go +++ b/pkg/cache/natskv.go @@ -51,7 +51,7 @@ func newNATSKV[T any](cfg *config) (*natsKVCache[T], error) { go c.watchReconnect(cfg.reconnCh) } - cfg.logger.Infow("cache: using NATS KV backend", "bucket", cfg.bucketName, "ttl", cfg.ttl) + cfg.logger.Infow("msg", "cache: using NATS KV backend", "bucket", cfg.bucketName, "ttl", cfg.ttl) return c, nil } @@ -79,9 +79,9 @@ func (c *natsKVCache[T]) initBucket() error { func (c *natsKVCache[T]) watchReconnect(ch <-chan struct{}) { for range ch { - c.logger.Infow("cache: NATS reconnected, reinitializing bucket", "bucket", c.bucket) + c.logger.Infow("msg", "cache: NATS reconnected, reinitializing bucket", "bucket", c.bucket) if err := c.initBucket(); err != nil { - c.logger.Warnw("cache: failed to reinitialize bucket after reconnect", "bucket", c.bucket, "error", err) + c.logger.Warnw("msg", "cache: failed to reinitialize bucket after reconnect", "bucket", c.bucket, "error", err) } } } @@ -107,7 +107,7 @@ func (c *natsKVCache[T]) Get(ctx context.Context, key string) (T, bool, error) { var zero T kv := c.getKV() if kv == nil { - c.logger.Warnw("cache get: KV handle is nil, returning miss", "key", key, "backend", "nats") + c.logger.Warnw("msg", "cache get: KV handle is nil, returning miss", "key", key, "backend", "nats") return zero, false, nil } @@ -115,28 +115,28 @@ func (c *natsKVCache[T]) Get(ctx context.Context, key string) (T, bool, error) { entry, err := kv.Get(ctx, sKey) if err != nil { if errors.Is(err, jetstream.ErrKeyNotFound) { - c.logger.Debugw("cache get", "key", key, "hit", false, "backend", "nats") + c.logger.Debugw("msg", "cache get", "key", key, "hit", false, "backend", "nats") return zero, false, nil } - c.logger.Warnw("cache get error", "key", key, "error", err, "backend", "nats") + c.logger.Warnw("msg", "cache get error", "key", key, "error", err, "backend", "nats") return zero, false, nil } var val T if err := json.Unmarshal(entry.Value(), &val); err != nil { - c.logger.Warnw("cache get: unmarshal failed, deleting corrupted entry", "key", key, "error", err, "backend", "nats") + c.logger.Warnw("msg", "cache get: unmarshal failed, deleting corrupted entry", "key", key, "error", err, "backend", "nats") _ = kv.Delete(ctx, sKey) return zero, false, nil } - c.logger.Debugw("cache get", "key", key, "hit", true, "backend", "nats") + c.logger.Debugw("msg", "cache get", "key", key, "hit", true, "backend", "nats") return val, true, nil } func (c *natsKVCache[T]) Set(ctx context.Context, key string, value T) error { kv := c.getKV() if kv == nil { - c.logger.Warnw("cache set: KV handle is nil, skipping", "key", key, "backend", "nats") + c.logger.Warnw("msg", "cache set: KV handle is nil, skipping", "key", key, "backend", "nats") return nil } @@ -146,11 +146,11 @@ func (c *natsKVCache[T]) Set(ctx context.Context, key string, value T) error { } if _, err := kv.Put(ctx, sanitizeKey(key), data); err != nil { - c.logger.Warnw("cache set error", "key", key, "error", err, "backend", "nats") + c.logger.Warnw("msg", "cache set error", "key", key, "error", err, "backend", "nats") return nil } - c.logger.Debugw("cache set", "key", key, "backend", "nats") + c.logger.Debugw("msg", "cache set", "key", key, "backend", "nats") return nil } @@ -162,11 +162,11 @@ func (c *natsKVCache[T]) Delete(ctx context.Context, key string) error { if err := kv.Delete(ctx, sanitizeKey(key)); err != nil { if !errors.Is(err, jetstream.ErrKeyNotFound) { - c.logger.Warnw("cache delete error", "key", key, "error", err, "backend", "nats") + c.logger.Warnw("msg", "cache delete error", "key", key, "error", err, "backend", "nats") } } - c.logger.Debugw("cache delete", "key", key, "backend", "nats") + c.logger.Debugw("msg", "cache delete", "key", key, "backend", "nats") return nil } @@ -181,16 +181,16 @@ func (c *natsKVCache[T]) Purge(ctx context.Context) error { if errors.Is(err, jetstream.ErrNoKeysFound) { return nil } - c.logger.Warnw("cache purge: failed to list keys", "error", err, "backend", "nats") + c.logger.Warnw("msg", "cache purge: failed to list keys", "error", err, "backend", "nats") return nil } for _, k := range keys { if err := kv.Purge(ctx, k); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { - c.logger.Warnw("cache purge: failed to purge key", "key", k, "error", err, "backend", "nats") + c.logger.Warnw("msg", "cache purge: failed to purge key", "key", k, "error", err, "backend", "nats") } } - c.logger.Debugw("cache purge", "backend", "nats") + c.logger.Debugw("msg", "cache purge", "backend", "nats") return nil } diff --git a/pkg/cache/policyevalbundle/policyevalbundle.go b/pkg/cache/policyevalbundle/policyevalbundle.go new file mode 100644 index 000000000..fb7fa5d29 --- /dev/null +++ b/pkg/cache/policyevalbundle/policyevalbundle.go @@ -0,0 +1,60 @@ +// +// Copyright 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. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Package policyevalbundle provides a typed cache for policy evaluation bundles. +package policyevalbundle + +import ( + "context" + "time" + + "github.com/chainloop-dev/chainloop/pkg/cache" + "github.com/chainloop-dev/chainloop/pkg/natsconn" + "github.com/go-kratos/kratos/v2/log" +) + +const ( + ttl = 24 * time.Hour + bucket = "chainloop-policy-eval-bundles" + description = "Cache for policy evaluation bundles from CAS" +) + +// Cache wraps cache.Cache[[]byte] to provide a distinct type for wire disambiguation. +type Cache struct { + cache.Cache[[]byte] +} + +// New creates a policy evaluation bundle cache with built-in TTL, bucket, and description. +func New(ctx context.Context, rc *natsconn.ReloadableConnection, logger log.Logger) (*Cache, error) { + opts := []cache.Option{ + cache.WithTTL(ttl), + cache.WithDescription(description), + } + + if logger != nil { + opts = append(opts, cache.WithLogger(log.NewHelper(logger))) + } + + if rc != nil { + opts = append(opts, cache.WithNATS(rc.Conn, bucket)) + opts = append(opts, cache.WithReconnect(rc.Subscribe(ctx))) + } + + c, err := cache.New[[]byte](opts...) + if err != nil { + return nil, err + } + return &Cache{Cache: c}, nil +}