From 7aac671e9cc3368ad0274efff04d5e84cb117324 Mon Sep 17 00:00:00 2001 From: ilhom Date: Wed, 12 Aug 2026 13:34:48 +0700 Subject: [PATCH 1/2] fix(finance): exclude MB products from the generic calc engine MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MB costs are owned by MB Batch (MB Push to Head). Letting a calc job also compute them gave two writers for the same (product, period, calc_type): a calc job run after a push superseded the APPROVED cst_product_cost row, leaving cst_mb_cost pointing at a SUPERSEDED source with a stale value. Exclusion is layered: - Orchestrator (authoritative): allActiveProducts and productsByType filter MB out of the seed set; resolveInitialSet rejects SINGLE_PRODUCT on an MB product and FILTERED by the MB type. Under ALL the drop is silent — MB is simply not part of what the calc engine owns. loadProductRMEdges is deliberately left unfiltered: the rule constrains what a job targets, not how dependencies resolve, so yarn products consuming MB still price correctly. - Finance trigger handler: an optional MBTypeChecker guard turns what would be an async FAILED job into a synchronous 400 before any cal_job row is written. Wired via WithMBGuard so existing call sites are unchanged. Verified against dev: scope ALL seeds 9457 products vs 13625 before, a difference of exactly 4168 — the MB active-route count — and writes zero MB rows. Forced MB paths (FILTERED type 81, SINGLE_PRODUCT on an MB product) both return 400. Non-MB types are untouched. Co-Authored-By: Claude Opus 5 --- .../internal/orchestrator/dag_builder.go | 114 +++++++++++- .../internal/orchestrator/dag_builder_test.go | 130 ++++++++++++++ services/finance/cmd/server/main.go | 2 +- .../application/costcalc/trigger_handler.go | 91 +++++++++- .../costcalc/trigger_mb_guard_test.go | 170 ++++++++++++++++++ .../delivery/grpc/cost_calc_handler.go | 1 + .../postgres/mb_type_checker.go | 50 ++++++ 7 files changed, 546 insertions(+), 12 deletions(-) create mode 100644 services/finance/internal/application/costcalc/trigger_mb_guard_test.go create mode 100644 services/finance/internal/infrastructure/postgres/mb_type_checker.go diff --git a/services/finance-cost-orchestrator/internal/orchestrator/dag_builder.go b/services/finance-cost-orchestrator/internal/orchestrator/dag_builder.go index a3d95963..02c8a3e4 100644 --- a/services/finance-cost-orchestrator/internal/orchestrator/dag_builder.go +++ b/services/finance-cost-orchestrator/internal/orchestrator/dag_builder.go @@ -4,6 +4,7 @@ package orchestrator import ( "context" "database/sql" + "errors" "fmt" "github.com/lib/pq" @@ -82,6 +83,43 @@ type edge struct { upstream int64 // the product referenced as RM } +// typeCodeMB is the cost_product_type.cpt_type_code of Master Batch products. +// +// MB is computed exclusively by the finance MB_BATCH path (mbbatch.RunMBBatch, +// reached from the "MB Push to Head" page), which computes all three calc types in +// one dependency-ordered run and then pushes the results into cst_mb_cost. Letting +// the generic calc engine also compute MB gives two writers for the same +// (product, period, calc_type): a calc job supersedes the APPROVED cst_product_cost +// row that a push already consumed, leaving cst_mb_cost pointing at a SUPERSEDED +// source while its value stays stale. MB is therefore excluded from every calc-job +// scope — see excludeMBPredicate. +// +// Non-MB products are unaffected: MB is never referenced as a PRODUCT-type route RM +// by any other product type (only MB routes reference MB). Downstream consumers such +// as yarn read MB prices through cst_mb_cost via LoadMBCosts, a formula path that +// does not involve the route DAG at all. +const typeCodeMB = "MB" + +// excludeMBPredicate is the SQL fragment that drops MB-typed products from a product +// selection. It is a NOT EXISTS rather than a JOIN so it can be appended to queries +// that do not already join the product master, and so a product without a master row +// is kept rather than silently dropped. +const excludeMBPredicate = ` + AND NOT EXISTS ( + SELECT 1 + FROM cost_product_master mb_pm + JOIN cost_product_type mb_pt ON mb_pt.cpt_type_id = mb_pm.cpm_product_type_id + WHERE mb_pm.cpm_product_sys_id = %s + AND mb_pt.cpt_type_code = '` + typeCodeMB + `' + )` + +// ErrMBScopeNotAllowed is returned when a calc job explicitly targets MB products. +// Unlike the silent filtering applied to ALL, an explicit MB request is a user +// mistake worth surfacing: the job would otherwise report SUCCESS having calculated +// nothing. +var ErrMBScopeNotAllowed = errors.New( + "MB products are calculated by the MB Batch path (MB Push to Head), not by calc jobs") + // resolveInitialSet returns the product_sys_ids to start traversal from. func (b *DagBuilder) resolveInitialSet(ctx context.Context, in ScopeInput) ([]int64, error) { switch in.Scope { @@ -89,6 +127,13 @@ func (b *DagBuilder) resolveInitialSet(ctx context.Context, in ScopeInput) ([]in if in.ProductSysID == 0 { return nil, fmt.Errorf("product_sys_id required for SINGLE_PRODUCT") } + isMB, err := b.isMBProduct(ctx, in.ProductSysID) + if err != nil { + return nil, err + } + if isMB { + return nil, fmt.Errorf("product %d: %w", in.ProductSysID, ErrMBScopeNotAllowed) + } return []int64{in.ProductSysID}, nil case costcalc.ScopeSingleRoute: @@ -98,6 +143,13 @@ func (b *DagBuilder) resolveInitialSet(ctx context.Context, in ScopeInput) ([]in return b.productsOfRouteHead(ctx, in.RouteHeadID) case costcalc.ScopeFiltered: + isMB, err := b.isMBProductType(ctx, in.ProductTypeIDFilter) + if err != nil { + return nil, err + } + if isMB { + return nil, ErrMBScopeNotAllowed + } return b.productsByType(ctx, in.ProductTypeIDFilter) case costcalc.ScopeAll: @@ -107,41 +159,79 @@ func (b *DagBuilder) resolveInitialSet(ctx context.Context, in ScopeInput) ([]in } } -// allActiveProducts returns every product that has an active (COMPLETE or LOCKED) route head. +// allActiveProducts returns every product that has an active (COMPLETE or LOCKED) +// route head, excluding MB (see typeCodeMB). Under ALL the exclusion is silent — +// the user asked for "everything the calc engine owns", and MB is not part of that. func (b *DagBuilder) allActiveProducts(ctx context.Context) ([]int64, error) { - const q = ` + q := ` SELECT DISTINCT crh.crh_product_sys_id FROM cost_route_head crh WHERE crh.crh_routing_status IN ('COMPLETE','LOCKED') - AND crh.crh_deleted_at IS NULL + AND crh.crh_deleted_at IS NULL` + + fmt.Sprintf(excludeMBPredicate, "crh.crh_product_sys_id") + ` ORDER BY crh.crh_product_sys_id ` return b.scanInt64s(ctx, q) } -// productsByType returns active-route products of a specific product type. +// productsByType returns active-route products of a specific product type. The MB +// type itself is rejected earlier in resolveInitialSet; the predicate here is +// belt-and-braces for the case where two type rows share the MB code. func (b *DagBuilder) productsByType(ctx context.Context, typeID int32) ([]int64, error) { if typeID == 0 { return nil, fmt.Errorf("product_type_id_filter required for FILTERED scope") } - const q = ` + q := ` SELECT DISTINCT crh.crh_product_sys_id FROM cost_route_head crh JOIN cost_product_master cpm ON cpm.cpm_product_sys_id = crh.crh_product_sys_id WHERE crh.crh_routing_status IN ('COMPLETE','LOCKED') AND crh.crh_deleted_at IS NULL - AND cpm.cpm_product_type_id = $1 + AND cpm.cpm_product_type_id = $1` + + fmt.Sprintf(excludeMBPredicate, "crh.crh_product_sys_id") + ` ORDER BY crh.crh_product_sys_id ` return b.scanInt64s(ctx, q, typeID) } -// productsOfRouteHead returns the FG product for a specific route head. +// productsOfRouteHead returns the FG product for a specific route head, unless that +// product is an MB. func (b *DagBuilder) productsOfRouteHead(ctx context.Context, headID int64) ([]int64, error) { - const q = `SELECT crh_product_sys_id FROM cost_route_head WHERE crh_head_id = $1 AND crh_deleted_at IS NULL` + q := `SELECT crh_product_sys_id FROM cost_route_head + WHERE crh_head_id = $1 AND crh_deleted_at IS NULL` + + fmt.Sprintf(excludeMBPredicate, "crh_product_sys_id") return b.scanInt64s(ctx, q, headID) } +// isMBProduct reports whether a single product is MB-typed. +func (b *DagBuilder) isMBProduct(ctx context.Context, productSysID int64) (bool, error) { + const q = ` + SELECT EXISTS ( + SELECT 1 + FROM cost_product_master pm + JOIN cost_product_type pt ON pt.cpt_type_id = pm.cpm_product_type_id + WHERE pm.cpm_product_sys_id = $1 AND pt.cpt_type_code = $2 + )` + var isMB bool + if err := b.db.QueryRowContext(ctx, q, productSysID, typeCodeMB).Scan(&isMB); err != nil { + return false, fmt.Errorf("check MB product %d: %w", productSysID, err) + } + return isMB, nil +} + +// isMBProductType reports whether a product type id is the MB type. +func (b *DagBuilder) isMBProductType(ctx context.Context, typeID int32) (bool, error) { + if typeID == 0 { + return false, nil // productsByType raises the "filter required" error. + } + const q = `SELECT EXISTS (SELECT 1 FROM cost_product_type WHERE cpt_type_id = $1 AND cpt_type_code = $2)` + var isMB bool + if err := b.db.QueryRowContext(ctx, q, typeID, typeCodeMB).Scan(&isMB); err != nil { + return false, fmt.Errorf("check MB product type %d: %w", typeID, err) + } + return isMB, nil +} + // loadProductRMEdges returns the PRODUCT-type RM edges (downstream -> upstream) // for the given set of product_sys_ids. // @@ -152,6 +242,14 @@ func (b *DagBuilder) productsOfRouteHead(ctx context.Context, headID int64) ([]i // cost, so it must NOT become a graph node. Without this guard such a target // would be added as a headless node and later fail the cal_job_product insert // (cjp_route_head_id NOT NULL), aborting the whole job. +// +// Deliberately NOT filtered by typeCodeMB. The MB exclusion applies to what a job +// TARGETS, not to dependency resolution: if some non-MB product ever does reference +// an MB as a PRODUCT-type RM, that MB is a genuine upstream input and must still be +// computed, exactly as it is today, or the referencing product would fail with +// ErrMissingUpstreamCost. No such reference exists in current data — MB routes are +// the only routes that reference MB — so this branch is unreachable in practice and +// the seed-set filters above carry the whole exclusion. func (b *DagBuilder) loadProductRMEdges(ctx context.Context, productSysIDs []int64) ([]edge, error) { const q = ` SELECT crs.crs_product_sys_id, crm.crm_rm_product_sys_id diff --git a/services/finance-cost-orchestrator/internal/orchestrator/dag_builder_test.go b/services/finance-cost-orchestrator/internal/orchestrator/dag_builder_test.go index e9da75e2..c3a66d6a 100644 --- a/services/finance-cost-orchestrator/internal/orchestrator/dag_builder_test.go +++ b/services/finance-cost-orchestrator/internal/orchestrator/dag_builder_test.go @@ -143,3 +143,133 @@ func TestDagBuilder_Filtered_RequiresTypeID(t *testing.T) { _, _, err := b.Build(context.Background(), ScopeInput{Scope: costcalc.ScopeFiltered}) require.Error(t, err) } + +// mbTypeID returns the MB product type id, skipping when the dev DB has none. +func mbTypeID(t *testing.T, db *sql.DB) int32 { + t.Helper() + var id int32 + err := db.QueryRowContext(context.Background(), + `SELECT cpt_type_id FROM cost_product_type WHERE cpt_type_code = $1`, typeCodeMB).Scan(&id) + if err == sql.ErrNoRows { + t.Skip("no MB product type in dev DB; skip") + } + require.NoError(t, err) + return id +} + +// A FILTERED job naming the MB type is a user mistake, not an empty result: MB is owned +// by the MB Batch path. It must fail loudly rather than report SUCCESS having computed +// nothing. +func TestDagBuilder_FilteredByMBType_Rejected(t *testing.T) { + db := openTestDB(t) + defer func() { _ = db.Close() }() + b := NewDagBuilder(db) + + _, _, err := b.Build(context.Background(), ScopeInput{ + Scope: costcalc.ScopeFiltered, + ProductTypeIDFilter: mbTypeID(t, db), + }) + require.Error(t, err) + require.ErrorIs(t, err, ErrMBScopeNotAllowed) +} + +// Same rule when the MB is named directly by product id. +func TestDagBuilder_SingleProductMB_Rejected(t *testing.T) { + db := openTestDB(t) + defer func() { _ = db.Close() }() + + var mbProduct int64 + err := db.QueryRowContext(context.Background(), ` + SELECT pm.cpm_product_sys_id + FROM cost_product_master pm + JOIN cost_product_type pt ON pt.cpt_type_id = pm.cpm_product_type_id + WHERE pt.cpt_type_code = $1 + LIMIT 1`, typeCodeMB).Scan(&mbProduct) + if err == sql.ErrNoRows { + t.Skip("no MB products in dev DB; skip") + } + require.NoError(t, err) + + b := NewDagBuilder(db) + _, _, err = b.Build(context.Background(), ScopeInput{ + Scope: costcalc.ScopeSingleProduct, + ProductSysID: mbProduct, + }) + require.Error(t, err) + require.ErrorIs(t, err, ErrMBScopeNotAllowed) +} + +// ALL excludes MB silently — the user asked for the calc engine's own population, which +// does not include MB. This is the assertion that the exclusion predicate actually bites. +func TestDagBuilder_All_ExcludesMBFromSeedSet(t *testing.T) { + db := openTestDB(t) + defer func() { _ = db.Close() }() + + seed, err := NewDagBuilder(db).allActiveProducts(context.Background()) + require.NoError(t, err) + if len(seed) == 0 { + t.Skip("no active routes in dev DB") + } + + rows, err := db.QueryContext(context.Background(), ` + SELECT pm.cpm_product_sys_id + FROM cost_product_master pm + JOIN cost_product_type pt ON pt.cpt_type_id = pm.cpm_product_type_id + WHERE pt.cpt_type_code = $1`, typeCodeMB) + require.NoError(t, err) + defer func() { _ = rows.Close() }() + + mb := map[int64]bool{} + for rows.Next() { + var id int64 + require.NoError(t, rows.Scan(&id)) + mb[id] = true + } + require.NoError(t, rows.Err()) + if len(mb) == 0 { + t.Skip("no MB products in dev DB; nothing to exclude") + } + + for _, id := range seed { + require.False(t, mb[id], "MB product %d must not be in the ScopeAll seed set", id) + } +} + +// The other product types must be entirely unaffected — this is the guarantee that the MB +// exclusion does not disturb the yarn path. Every non-MB product with an active route is +// still selected. +func TestDagBuilder_All_KeepsEveryNonMBProduct(t *testing.T) { + db := openTestDB(t) + defer func() { _ = db.Close() }() + + seed, err := NewDagBuilder(db).allActiveProducts(context.Background()) + require.NoError(t, err) + if len(seed) == 0 { + t.Skip("no active routes in dev DB") + } + got := map[int64]bool{} + for _, id := range seed { + got[id] = true + } + + rows, err := db.QueryContext(context.Background(), ` + SELECT DISTINCT crh.crh_product_sys_id + FROM cost_route_head crh + WHERE crh.crh_routing_status IN ('COMPLETE','LOCKED') + AND crh.crh_deleted_at IS NULL + AND NOT EXISTS ( + SELECT 1 FROM cost_product_master pm + JOIN cost_product_type pt ON pt.cpt_type_id = pm.cpm_product_type_id + WHERE pm.cpm_product_sys_id = crh.crh_product_sys_id + AND pt.cpt_type_code = $1 + )`, typeCodeMB) + require.NoError(t, err) + defer func() { _ = rows.Close() }() + + for rows.Next() { + var id int64 + require.NoError(t, rows.Scan(&id)) + require.True(t, got[id], "non-MB product %d with an active route must still be selected", id) + } + require.NoError(t, rows.Err()) +} diff --git a/services/finance/cmd/server/main.go b/services/finance/cmd/server/main.go index 5a2fbdcf..9ccfd659 100644 --- a/services/finance/cmd/server/main.go +++ b/services/finance/cmd/server/main.go @@ -566,7 +566,7 @@ func run() error { //nolint:gocognit,gocyclo // linear service wiring / DI setup ) costCalcHandler := grpcdelivery.NewCostCalcHandler( calcSvc, - costcalc.NewTriggerJobHandler(calcSvc), + costcalc.NewTriggerJobHandler(calcSvc, costcalc.WithMBGuard(postgres.NewMBTypeChecker(db))), costcalc.NewGetJobHandler(calcSvc), costcalc.NewListJobsHandler(calcSvc), costcalc.NewListChunksHandler(calcSvc), diff --git a/services/finance/internal/application/costcalc/trigger_handler.go b/services/finance/internal/application/costcalc/trigger_handler.go index dc545d7c..b24f1c45 100644 --- a/services/finance/internal/application/costcalc/trigger_handler.go +++ b/services/finance/internal/application/costcalc/trigger_handler.go @@ -28,12 +28,42 @@ type TriggerCommand struct { // engine. Other scopes will land in S8c when the orchestrator + RMQ worker // machinery exists. type TriggerJobHandler struct { - svc *Service + svc *Service + mbGuard MBTypeChecker +} + +// MBTypeChecker answers whether a product / product type is Master Batch, so the +// trigger path can reject MB calc jobs before a cal_job row is even created. +// +// MB is owned by the MB_BATCH path (mbbatch.RunMBBatch, "MB Push to Head"), which +// computes ACTUAL/SELLING/FORECAST in one dependency-ordered run and pushes the +// results into cst_mb_cost. A calc job over the same product would supersede the +// APPROVED cst_product_cost row that a push already consumed, leaving cst_mb_cost +// pointing at a SUPERSEDED source with a stale value. The orchestrator enforces the +// same rule on its side (see its dag_builder.go); this check exists so the user gets +// a synchronous error instead of an asynchronously FAILED job. +type MBTypeChecker interface { + IsMBProduct(ctx context.Context, productSysID int64) (bool, error) + IsMBProductType(ctx context.Context, productTypeID int32) (bool, error) +} + +// TriggerOption customizes the handler at construction. +type TriggerOption func(*TriggerJobHandler) + +// WithMBGuard installs the MB rejection check. Omitting it leaves the finance-side +// check disabled — the orchestrator still enforces the rule, so behavior stays +// correct, just less immediate. Tests omit it. +func WithMBGuard(c MBTypeChecker) TriggerOption { + return func(h *TriggerJobHandler) { h.mbGuard = c } } // NewTriggerJobHandler constructs the handler. -func NewTriggerJobHandler(svc *Service) *TriggerJobHandler { - return &TriggerJobHandler{svc: svc} +func NewTriggerJobHandler(svc *Service, opts ...TriggerOption) *TriggerJobHandler { + h := &TriggerJobHandler{svc: svc} + for _, opt := range opts { + opt(h) + } + return h } // auditEntityKindJob is the EntityKind value for all COST_CALC_JOB_* audit events. @@ -47,6 +77,12 @@ var ErrScopeNotYetSupported = errors.New("scope not yet supported in S8b foundat // product id. var ErrProductRequired = errors.New("product_sys_id required for SINGLE_PRODUCT scope") +// ErrMBNotCalcJobEligible is returned when a calc job explicitly targets Master +// Batch products. MB is computed by MB Batch (MB Push to Head), which owns all +// three calc types; see MBTypeChecker for why a second writer is harmful. +var ErrMBNotCalcJobEligible = errors.New( + "MB products are calculated from MB Batch (MB Push to Head), not from calc jobs") + // Handle creates a job + chunk + job_product row, runs ProcessChunk inline, // finalizes the job, and returns the fully-resolved Job aggregate. // @@ -57,6 +93,9 @@ var ErrProductRequired = errors.New("product_sys_id required for SINGLE_PRODUCT // there too — the orchestrator walks the full upstream DAG and computes // intermediates first. Only fall back to the inline path when RMQ is offline. func (h *TriggerJobHandler) Handle(ctx context.Context, cmd TriggerCommand) (*costcalcdom.Job, error) { + if err := h.rejectMBScope(ctx, cmd); err != nil { + return nil, err + } if cmd.Scope != costcalcdom.ScopeSingleProduct { return h.dispatchToOrchestrator(ctx, cmd) } @@ -218,6 +257,52 @@ func (h *TriggerJobHandler) completeJob(ctx context.Context, job *costcalcdom.Jo return nil } +// rejectMBScope fails a trigger that explicitly targets MB, before any cal_job row +// is written. Only the two scopes that can NAME MB are checked: +// +// - SINGLE_PRODUCT, when the product itself is MB; +// - FILTERED, when the chosen product type is MB. +// +// ALL is deliberately not rejected — the user is asking for the calc engine's whole +// population, which simply does not include MB. The orchestrator filters MB out of +// that seed set silently. +// +// SINGLE_ROUTE is left to the orchestrator: resolving a route head to its product +// needs a query this handler has no repository for, and the orchestrator's +// productsOfRouteHead already drops MB. +func (h *TriggerJobHandler) rejectMBScope(ctx context.Context, cmd TriggerCommand) error { + if h.mbGuard == nil { + return nil + } + switch cmd.Scope { + case costcalcdom.ScopeSingleProduct: + if cmd.ProductSysID == 0 { + return nil // ErrProductRequired is raised by the caller. + } + isMB, err := h.mbGuard.IsMBProduct(ctx, cmd.ProductSysID) + if err != nil { + return fmt.Errorf("check MB product: %w", err) + } + if isMB { + return fmt.Errorf("product %d: %w", cmd.ProductSysID, ErrMBNotCalcJobEligible) + } + case costcalcdom.ScopeFiltered: + if cmd.ProductTypeIDFilter == 0 { + return nil + } + isMB, err := h.mbGuard.IsMBProductType(ctx, cmd.ProductTypeIDFilter) + if err != nil { + return fmt.Errorf("check MB product type: %w", err) + } + if isMB { + return ErrMBNotCalcJobEligible + } + case costcalcdom.ScopeAll, costcalcdom.ScopeSingleRoute, costcalcdom.ScopeMBBatch: + return nil + } + return nil +} + // dispatchToOrchestrator inserts a QUEUED cal_job row + publishes a // JobTriggeredEvent so the orchestrator picks up planning + execution. The // returned Job is in QUEUED state — the orchestrator drives all further diff --git a/services/finance/internal/application/costcalc/trigger_mb_guard_test.go b/services/finance/internal/application/costcalc/trigger_mb_guard_test.go new file mode 100644 index 00000000..4b672e03 --- /dev/null +++ b/services/finance/internal/application/costcalc/trigger_mb_guard_test.go @@ -0,0 +1,170 @@ +package costcalc + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/require" + + costcalcdom "github.com/mutugading/goapps-backend/services/finance/internal/domain/costcalc" +) + +// stubMBChecker records what it was asked and answers from fixed values. +type stubMBChecker struct { + productIsMB bool + typeIsMB bool + err error + askedProduct int64 + askedType int32 + productCalls int + typeCalls int +} + +func (s *stubMBChecker) IsMBProduct(_ context.Context, productSysID int64) (bool, error) { + s.productCalls++ + s.askedProduct = productSysID + return s.productIsMB, s.err +} + +func (s *stubMBChecker) IsMBProductType(_ context.Context, productTypeID int32) (bool, error) { + s.typeCalls++ + s.askedType = productTypeID + return s.typeIsMB, s.err +} + +// A deliberate FILTERED-by-MB-type trigger must fail before any cal_job row is written, +// because MB is owned by the MB_BATCH path. Service is empty: reaching any repo would +// nil-panic, which is exactly the assertion that nothing was persisted. +func TestTriggerJob_FilteredByMBType_Rejected(t *testing.T) { + t.Parallel() + guard := &stubMBChecker{typeIsMB: true} + h := NewTriggerJobHandler(&Service{}, WithMBGuard(guard)) + + _, err := h.Handle(context.Background(), TriggerCommand{ + Period: "202607", + CalcType: costcalcdom.CalcTypeActual, + Scope: costcalcdom.ScopeFiltered, + ProductTypeIDFilter: 81, + Actor: "test", + TriggeredBy: "TEST", + }) + require.Error(t, err) + require.True(t, errors.Is(err, ErrMBNotCalcJobEligible)) + require.Equal(t, int32(81), guard.askedType) +} + +// A non-MB product type is untouched by the guard: the request proceeds and fails only +// later, for the unrelated reason that no orchestrator publisher is configured. +func TestTriggerJob_FilteredByNonMBType_PassesGuard(t *testing.T) { + t.Parallel() + guard := &stubMBChecker{typeIsMB: false} + h := NewTriggerJobHandler(&Service{}, WithMBGuard(guard)) + + _, err := h.Handle(context.Background(), TriggerCommand{ + Period: "202607", + CalcType: costcalcdom.CalcTypeActual, + Scope: costcalcdom.ScopeFiltered, + ProductTypeIDFilter: 12, + Actor: "test", + TriggeredBy: "TEST", + }) + require.Error(t, err) + require.False(t, errors.Is(err, ErrMBNotCalcJobEligible)) + require.True(t, errors.Is(err, ErrScopeNotYetSupported)) +} + +func TestTriggerJob_SingleProductMB_Rejected(t *testing.T) { + t.Parallel() + guard := &stubMBChecker{productIsMB: true} + h := NewTriggerJobHandler(&Service{}, WithMBGuard(guard)) + + _, err := h.Handle(context.Background(), TriggerCommand{ + Period: "202607", + CalcType: costcalcdom.CalcTypeActual, + Scope: costcalcdom.ScopeSingleProduct, + ProductSysID: 40418, + Actor: "test", + TriggeredBy: "TEST", + }) + require.Error(t, err) + require.True(t, errors.Is(err, ErrMBNotCalcJobEligible)) + require.Equal(t, int64(40418), guard.askedProduct) +} + +// ALL is the "everything the calc engine owns" request. MB simply is not part of that +// population, so the trigger is not rejected — the orchestrator filters MB out of the +// seed set silently. Asserting the guard is never consulted keeps that contract honest. +func TestTriggerJob_ScopeAll_NotRejectedAndGuardNotConsulted(t *testing.T) { + t.Parallel() + guard := &stubMBChecker{productIsMB: true, typeIsMB: true} + h := NewTriggerJobHandler(&Service{}, WithMBGuard(guard)) + + _, err := h.Handle(context.Background(), TriggerCommand{ + Period: "202607", + CalcType: costcalcdom.CalcTypeActual, + Scope: costcalcdom.ScopeAll, + Actor: "test", + TriggeredBy: "TEST", + }) + require.Error(t, err) + require.False(t, errors.Is(err, ErrMBNotCalcJobEligible)) + require.Zero(t, guard.productCalls) + require.Zero(t, guard.typeCalls) +} + +// FILTERED with no type chosen has nothing to check; the guard must not be consulted and +// must not invent a rejection. +func TestTriggerJob_FilteredWithoutType_SkipsGuard(t *testing.T) { + t.Parallel() + guard := &stubMBChecker{typeIsMB: true} + h := NewTriggerJobHandler(&Service{}, WithMBGuard(guard)) + + _, err := h.Handle(context.Background(), TriggerCommand{ + Period: "202607", + CalcType: costcalcdom.CalcTypeActual, + Scope: costcalcdom.ScopeFiltered, + Actor: "test", + TriggeredBy: "TEST", + }) + require.False(t, errors.Is(err, ErrMBNotCalcJobEligible)) + require.Zero(t, guard.typeCalls) +} + +// Without WithMBGuard the finance-side check is disabled entirely (the orchestrator still +// enforces the rule). Existing constructor call sites must keep working unchanged. +func TestTriggerJob_NilGuard_NoRejection(t *testing.T) { + t.Parallel() + h := NewTriggerJobHandler(&Service{}) + + _, err := h.Handle(context.Background(), TriggerCommand{ + Period: "202607", + CalcType: costcalcdom.CalcTypeActual, + Scope: costcalcdom.ScopeFiltered, + ProductTypeIDFilter: 81, + Actor: "test", + TriggeredBy: "TEST", + }) + require.Error(t, err) + require.False(t, errors.Is(err, ErrMBNotCalcJobEligible)) +} + +// A checker failure must not be swallowed into "not MB": a job that should have been +// rejected would otherwise be created. +func TestTriggerJob_GuardError_Propagates(t *testing.T) { + t.Parallel() + sentinel := errors.New("db down") + guard := &stubMBChecker{err: sentinel} + h := NewTriggerJobHandler(&Service{}, WithMBGuard(guard)) + + _, err := h.Handle(context.Background(), TriggerCommand{ + Period: "202607", + CalcType: costcalcdom.CalcTypeActual, + Scope: costcalcdom.ScopeSingleProduct, + ProductSysID: 40418, + Actor: "test", + TriggeredBy: "TEST", + }) + require.Error(t, err) + require.True(t, errors.Is(err, sentinel)) +} diff --git a/services/finance/internal/delivery/grpc/cost_calc_handler.go b/services/finance/internal/delivery/grpc/cost_calc_handler.go index 61eb0bd0..b5ca1ed0 100644 --- a/services/finance/internal/delivery/grpc/cost_calc_handler.go +++ b/services/finance/internal/delivery/grpc/cost_calc_handler.go @@ -1238,6 +1238,7 @@ func mappedCostCalcErrToBase(err error) *commonv1.BaseResponse { case errors.Is(err, costcalc.ErrScopeNotYetSupported): return ErrorResponse("501", err.Error()) case errors.Is(err, costcalc.ErrProductRequired), + errors.Is(err, costcalc.ErrMBNotCalcJobEligible), errors.Is(err, costcalcdom.ErrInvalidPeriod): return ErrorResponse("400", err.Error()) // Configuration gap, not a server fault: Finance has not entered a diff --git a/services/finance/internal/infrastructure/postgres/mb_type_checker.go b/services/finance/internal/infrastructure/postgres/mb_type_checker.go new file mode 100644 index 00000000..bfdc0839 --- /dev/null +++ b/services/finance/internal/infrastructure/postgres/mb_type_checker.go @@ -0,0 +1,50 @@ +package postgres + +import ( + "context" + "fmt" +) + +// MBTypeChecker answers MB-typed questions for the calc-job trigger guard. It +// implements costcalc.MBTypeChecker. +// +// MB costs are produced exclusively by the MB_BATCH path ("MB Push to Head"), which +// computes ACTUAL/SELLING/FORECAST in one dependency-ordered run and pushes the results +// into cst_mb_cost. A generic calc job over the same product would be a second writer +// for the same (product, period, calc_type) and would supersede the APPROVED +// cst_product_cost row a push already consumed. +type MBTypeChecker struct { + db *DB +} + +// NewMBTypeChecker constructs the checker. +func NewMBTypeChecker(db *DB) *MBTypeChecker { + return &MBTypeChecker{db: db} +} + +// IsMBProduct reports whether the product is of the MB product type. A product with no +// master row is reported as not MB, matching the orchestrator's NOT EXISTS semantics. +func (c *MBTypeChecker) IsMBProduct(ctx context.Context, productSysID int64) (bool, error) { + const q = ` + SELECT EXISTS ( + SELECT 1 + FROM cost_product_master pm + JOIN cost_product_type pt ON pt.cpt_type_id = pm.cpm_product_type_id + WHERE pm.cpm_product_sys_id = $1 AND pt.cpt_type_code = $2 + )` + var isMB bool + if err := c.db.QueryRowContext(ctx, q, productSysID, mbCostProductTypeCode).Scan(&isMB); err != nil { + return false, fmt.Errorf("check MB product %d: %w", productSysID, err) + } + return isMB, nil +} + +// IsMBProductType reports whether the product type id is the MB type. +func (c *MBTypeChecker) IsMBProductType(ctx context.Context, productTypeID int32) (bool, error) { + const q = `SELECT EXISTS (SELECT 1 FROM cost_product_type WHERE cpt_type_id = $1 AND cpt_type_code = $2)` + var isMB bool + if err := c.db.QueryRowContext(ctx, q, productTypeID, mbCostProductTypeCode).Scan(&isMB); err != nil { + return false, fmt.Errorf("check MB product type %d: %w", productTypeID, err) + } + return isMB, nil +} From 1a4f6760de1f6f623f47c24359cf1dd09d1fd4fd Mon Sep 17 00:00:00 2001 From: ilhom Date: Wed, 12 Aug 2026 14:43:50 +0700 Subject: [PATCH 2/2] feat(finance): flag stale MB pushes and close two MB guard gaps MB products are computed exclusively by MB Batch and approved via push-to-head. Three gaps let that invariant erode silently. 1. needs-repush visibility (mbpush) Re-running MB Batch after a push supersedes the approved source row and inserts a fresh CALCULATED one, leaving cst_mb_cost pointing at a superseded row. The value is not wrong -- it is the last approved one -- but nothing told the operator a newer result was waiting. Preview now flags such heads and reports needs_repush_count. They remain pushable; re-pushing is the remedy. Classification is read-only: no bucket moves, no auto-push, no write-path change. 2. manual verify/approve accepted MB rows (costcalc) VerifyCost/ApproveCost took any cost_id. Flipping an MB row to VERIFIED made push-to-head skip it, since push requires status exactly CALCULATED. Both now reject MB-owned rows with a 400. MarkApprovedFromCalculatedTx -- the legitimate push path -- is a distinct method and stays unblocked. 3. calc engine could persist an MB dependency node (costcalc) loadProductRMEdges is deliberately unfiltered so a parent consuming an MB still resolves it; that left ProcessChunk able to supersede an MB row pulled in as a dependency. Closed at the write site: such products are marked BLOCKED with reason MB_OWNED_BY_MB_BATCH instead of persisted. Parents read committed state via LoadUpstreamCosts, so skipping the persist changes no parent's number. Non-MB products are unaffected: every guard keys on a single cpt_type_code = 'MB' equality and degrades to a no-op when unwired. Verified against live data -- all 350 PRODUCT edges to an MB come from another MB, zero from non-MB. Tests: finance and finance-cost-orchestrator go test -race pass, including TestDagBuilder_All_KeepsEveryNonMBProduct, TestProcessChunk_AllNonMB_EveryProductPersisted, and the non-MB pass-through cases for both guards. golangci-lint v2.3.0: 0 issues. Co-Authored-By: Claude Opus 5 --- gen/finance/v1/yarn_master.pb.go | 40 ++- .../finance/v1/yarn_master.swagger.json | 9 + services/finance/cmd/server/main.go | 12 +- .../costcalc/approve_cost_handler.go | 23 +- .../application/costcalc/mb_cost_row_guard.go | 48 ++++ .../costcalc/mb_cost_row_guard_test.go | 137 +++++++++++ .../application/costcalc/process_chunk.go | 49 ++++ .../costcalc/process_chunk_mb_guard_test.go | 231 ++++++++++++++++++ .../internal/application/costcalc/service.go | 24 +- .../costcalc/verify_cost_handler.go | 23 +- .../internal/application/mbpush/port.go | 8 + .../application/mbpush/preview_handler.go | 68 +++++- .../mbpush/preview_handler_test.go | 203 +++++++++++++++ .../delivery/grpc/cost_calc_handler.go | 1 + .../internal/delivery/grpc/mb_push_handler.go | 18 +- .../postgres/cst_mb_cost_repository.go | 45 ++++ .../postgres/mb_type_checker.go | 58 +++++ 17 files changed, 957 insertions(+), 40 deletions(-) create mode 100644 services/finance/internal/application/costcalc/mb_cost_row_guard.go create mode 100644 services/finance/internal/application/costcalc/mb_cost_row_guard_test.go create mode 100644 services/finance/internal/application/costcalc/process_chunk_mb_guard_test.go create mode 100644 services/finance/internal/application/mbpush/preview_handler_test.go diff --git a/gen/finance/v1/yarn_master.pb.go b/gen/finance/v1/yarn_master.pb.go index 5c51ef66..0e1d4088 100644 --- a/gen/finance/v1/yarn_master.pb.go +++ b/gen/finance/v1/yarn_master.pb.go @@ -14838,7 +14838,12 @@ type PushableMbHead struct { // Whether a SELLING cost value is available to push. HasSelling bool `protobuf:"varint,5,opt,name=has_selling,json=hasSelling,proto3" json:"has_selling,omitempty"` // Whether a FORECAST cost value is available to push. - HasForecast bool `protobuf:"varint,6,opt,name=has_forecast,json=hasForecast,proto3" json:"has_forecast,omitempty"` + HasForecast bool `protobuf:"varint,6,opt,name=has_forecast,json=hasForecast,proto3" json:"has_forecast,omitempty"` + // Whether this head was already pushed for the period but its pushed cost has since gone + // stale — the source cst_product_cost row is SUPERSEDED (or unlinked) and a newer + // non-superseded row exists. Informational only: such a head is still pushable, and + // re-pushing is exactly the remedy. + NeedsRepush bool `protobuf:"varint,7,opt,name=needs_repush,json=needsRepush,proto3" json:"needs_repush,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -14915,6 +14920,13 @@ func (x *PushableMbHead) GetHasForecast() bool { return false } +func (x *PushableMbHead) GetNeedsRepush() bool { + if x != nil { + return x.NeedsRepush + } + return false +} + // SkippedMbHead describes an MB Head excluded from the push, with the reason why. type SkippedMbHead struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -14996,9 +15008,12 @@ type PreviewPushToHeadResponse struct { // MB Heads eligible for push. Pushable []*PushableMbHead `protobuf:"bytes,2,rep,name=pushable,proto3" json:"pushable,omitempty"` // MB Heads excluded from push, with reasons. - Skipped []*SkippedMbHead `protobuf:"bytes,3,rep,name=skipped,proto3" json:"skipped,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + Skipped []*SkippedMbHead `protobuf:"bytes,3,rep,name=skipped,proto3" json:"skipped,omitempty"` + // Count of pushable heads flagged needs_repush — heads whose already-pushed cost went stale + // after a later MB Batch run. Surfaced as its own bucket so operators can act on it. + NeedsRepushCount int32 `protobuf:"varint,4,opt,name=needs_repush_count,json=needsRepushCount,proto3" json:"needs_repush_count,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *PreviewPushToHeadResponse) Reset() { @@ -15052,6 +15067,13 @@ func (x *PreviewPushToHeadResponse) GetSkipped() []*SkippedMbHead { return nil } +func (x *PreviewPushToHeadResponse) GetNeedsRepushCount() int32 { + if x != nil { + return x.NeedsRepushCount + } + return 0 +} + // ExecutePushToHeadRequest is the request for executing a push-to-head batch for a period. type ExecutePushToHeadRequest struct { state protoimpl.MessageState `protogen:"open.v1"` @@ -17013,7 +17035,7 @@ const file_finance_v1_yarn_master_proto_rawDesc = "" + "\x04base\x18\x01 \x01(\v2\x17.common.v1.BaseResponseR\x04base\"E\n" + "\x18PreviewPushToHeadRequest\x12)\n" + "\x06period\x18\x01 \x01(\tB\x11\xbaH\x0er\f2\n" + - "^[0-9]{6}$R\x06period\"\xb2\x01\n" + + "^[0-9]{6}$R\x06period\"\xd5\x01\n" + "\x0ePushableMbHead\x12\x15\n" + "\x06mbh_id\x18\x01 \x01(\tR\x05mbhId\x12\x12\n" + "\x04code\x18\x02 \x01(\tR\x04code\x12\x12\n" + @@ -17022,16 +17044,18 @@ const file_finance_v1_yarn_master_proto_rawDesc = "" + "has_actual\x18\x04 \x01(\bR\thasActual\x12\x1f\n" + "\vhas_selling\x18\x05 \x01(\bR\n" + "hasSelling\x12!\n" + - "\fhas_forecast\x18\x06 \x01(\bR\vhasForecast\"f\n" + + "\fhas_forecast\x18\x06 \x01(\bR\vhasForecast\x12!\n" + + "\fneeds_repush\x18\a \x01(\bR\vneedsRepush\"f\n" + "\rSkippedMbHead\x12\x15\n" + "\x06mbh_id\x18\x01 \x01(\tR\x05mbhId\x12\x12\n" + "\x04code\x18\x02 \x01(\tR\x04code\x12\x12\n" + "\x04name\x18\x03 \x01(\tR\x04name\x12\x16\n" + - "\x06reason\x18\x04 \x01(\tR\x06reason\"\xb5\x01\n" + + "\x06reason\x18\x04 \x01(\tR\x06reason\"\xe3\x01\n" + "\x19PreviewPushToHeadResponse\x12+\n" + "\x04base\x18\x01 \x01(\v2\x17.common.v1.BaseResponseR\x04base\x126\n" + "\bpushable\x18\x02 \x03(\v2\x1a.finance.v1.PushableMbHeadR\bpushable\x123\n" + - "\askipped\x18\x03 \x03(\v2\x19.finance.v1.SkippedMbHeadR\askipped\"o\n" + + "\askipped\x18\x03 \x03(\v2\x19.finance.v1.SkippedMbHeadR\askipped\x12,\n" + + "\x12needs_repush_count\x18\x04 \x01(\x05R\x10needsRepushCount\"o\n" + "\x18ExecutePushToHeadRequest\x12)\n" + "\x06period\x18\x01 \x01(\tB\x11\xbaH\x0er\f2\n" + "^[0-9]{6}$R\x06period\x12(\n" + diff --git a/gen/openapi/finance/v1/yarn_master.swagger.json b/gen/openapi/finance/v1/yarn_master.swagger.json index cada0241..74e5dd57 100644 --- a/gen/openapi/finance/v1/yarn_master.swagger.json +++ b/gen/openapi/finance/v1/yarn_master.swagger.json @@ -7502,6 +7502,11 @@ "$ref": "#/definitions/v1SkippedMbHead" }, "description": "MB Heads excluded from push, with reasons." + }, + "needsRepushCount": { + "type": "integer", + "format": "int32", + "description": "Count of pushable heads flagged needs_repush — heads whose already-pushed cost went stale\nafter a later MB Batch run. Surfaced as its own bucket so operators can act on it." } }, "description": "PreviewPushToHeadResponse is the response listing pushable and skipped MB Heads for a period." @@ -7609,6 +7614,10 @@ "hasForecast": { "type": "boolean", "description": "Whether a FORECAST cost value is available to push." + }, + "needsRepush": { + "type": "boolean", + "description": "Whether this head was already pushed for the period but its pushed cost has since gone\nstale — the source cst_product_cost row is SUPERSEDED (or unlinked) and a newer\nnon-superseded row exists. Informational only: such a head is still pushable, and\nre-pushing is exactly the remedy." } }, "description": "PushableMbHead describes an MB Head eligible to receive pushed costs for the previewed period." diff --git a/services/finance/cmd/server/main.go b/services/finance/cmd/server/main.go index 9ccfd659..05f5ff98 100644 --- a/services/finance/cmd/server/main.go +++ b/services/finance/cmd/server/main.go @@ -560,13 +560,17 @@ func run() error { //nolint:gocognit,gocyclo // linear service wiring / DI setup costAuditHistoryRepo := postgres.NewCostAuditHistoryRepository(db) calcEvalCache := evaluator.NewCache() calcLoader := costcalc.NewProductLoader(db.DB) + // One checker instance serves all three MB guards: the trigger-time scope rejection, + // the persist-time refusal inside ProcessChunk, and the manual verify/approve block. + mbTypeChecker := postgres.NewMBTypeChecker(db) calcSvc := costcalc.NewService( calcJobRepo, calcChunkRepo, calcJobProductRepo, costResultRepo, costAuditHistoryRepo, calcLoader, calcEvalCache, nil, costCalcJobTriggerPub, + costcalc.WithMBProductGuard(mbTypeChecker), ) costCalcHandler := grpcdelivery.NewCostCalcHandler( calcSvc, - costcalc.NewTriggerJobHandler(calcSvc, costcalc.WithMBGuard(postgres.NewMBTypeChecker(db))), + costcalc.NewTriggerJobHandler(calcSvc, costcalc.WithMBGuard(mbTypeChecker)), costcalc.NewGetJobHandler(calcSvc), costcalc.NewListJobsHandler(calcSvc), costcalc.NewListChunksHandler(calcSvc), @@ -577,8 +581,8 @@ func run() error { //nolint:gocognit,gocyclo // linear service wiring / DI setup costcalc.NewListCostHistoryHandler(calcSvc), costcalc.NewListCostResultsHandler(calcSvc), costcalc.NewPeriodsHandler(calcSvc), - costcalc.NewVerifyCostHandler(calcSvc), - costcalc.NewApproveCostHandler(calcSvc), + costcalc.NewVerifyCostHandler(calcSvc, costcalc.WithVerifyMBGuard(mbTypeChecker)), + costcalc.NewApproveCostHandler(calcSvc, costcalc.WithApproveMBGuard(mbTypeChecker)), costcalc.NewGetRouteCostSheetHandler(calcSvc), costsheet.NewRequestExportHandler(jobRepo, costResultRepo, costSheetExportPublisher), costSheetExportURL, @@ -592,7 +596,7 @@ func run() error { //nolint:gocognit,gocyclo // linear service wiring / DI setup // MB Push-to-Head: adapts MBHeadRepository/CostResultRepository to mbpush's ports. mbPushHeadReader := mbpush.NewMBHeadReaderAdapter(mbHeadRepo) mbPushCostReader := mbpush.NewCostReaderAdapter(costResultRepo) - mbPushPreviewHandler := mbpush.NewPreviewHandler(mbPushHeadReader, mbPushCostReader) + mbPushPreviewHandler := mbpush.NewPreviewHandler(mbPushHeadReader, mbPushCostReader, cstMBCostRepo) mbPushExecuteHandler := mbpush.NewExecuteHandler(db, mbPushHeadReader, mbPushCostReader, cstMBCostRepo, mbPushLogRepo) mbPushHandler, err := grpcdelivery.NewMBPushHandler(mbPushPreviewHandler, mbPushExecuteHandler, mbPushLogRepo) if err != nil { diff --git a/services/finance/internal/application/costcalc/approve_cost_handler.go b/services/finance/internal/application/costcalc/approve_cost_handler.go index f5ff607d..ae19c351 100644 --- a/services/finance/internal/application/costcalc/approve_cost_handler.go +++ b/services/finance/internal/application/costcalc/approve_cost_handler.go @@ -14,12 +14,26 @@ type ApproveCostCommand struct { // ApproveCostHandler transitions a VERIFIED result to APPROVED. type ApproveCostHandler struct { - svc *Service + svc *Service + mbGuard MBCostRowChecker +} + +// ApproveOption customizes the handler at construction. +type ApproveOption func(*ApproveCostHandler) + +// WithApproveMBGuard installs the MB rejection check. Omitting it leaves the check +// disabled; tests omit it. +func WithApproveMBGuard(c MBCostRowChecker) ApproveOption { + return func(h *ApproveCostHandler) { h.mbGuard = c } } // NewApproveCostHandler constructs the handler. -func NewApproveCostHandler(svc *Service) *ApproveCostHandler { - return &ApproveCostHandler{svc: svc} +func NewApproveCostHandler(svc *Service, opts ...ApproveOption) *ApproveCostHandler { + h := &ApproveCostHandler{svc: svc} + for _, opt := range opts { + opt(h) + } + return h } // Handle executes the approval. @@ -30,6 +44,9 @@ func (h *ApproveCostHandler) Handle(ctx context.Context, cmd ApproveCostCommand) if cmd.Actor == "" { return errors.New(errMsgActorRequired) } + if err := rejectMBCostRow(ctx, h.mbGuard, cmd.CostID); err != nil { + return err + } if err := h.svc.resultRepo.MarkApproved(ctx, cmd.CostID, cmd.Actor); err != nil { return fmt.Errorf("mark approved: %w", err) } diff --git a/services/finance/internal/application/costcalc/mb_cost_row_guard.go b/services/finance/internal/application/costcalc/mb_cost_row_guard.go new file mode 100644 index 00000000..87f4dacb --- /dev/null +++ b/services/finance/internal/application/costcalc/mb_cost_row_guard.go @@ -0,0 +1,48 @@ +package costcalc + +import ( + "context" + "errors" + "fmt" +) + +// MBCostRowChecker answers whether a cst_product_cost row belongs to a Master Batch +// product, so the manual verify / approve RPCs can refuse it. +// +// MB rows are owned end-to-end by the MB_BATCH path: mbbatch writes them CALCULATED and +// MB Push-to-Head consumes them, flipping CALCULATED -> APPROVED inside its own +// transaction (CostResultRepository.MarkApprovedFromCalculatedTx) together with the +// cst_mb_cost upsert. Push requires the source row to be exactly CALCULATED. So a user +// hand-verifying an MB row through the generic Cost Results screen does not corrupt a +// number — it silently makes the next push SKIP that MB, which is worse: the MB keeps +// its stale cst_mb_cost value with no error anywhere. +// +// The guard blocks only the two manual RPC handlers. MarkApprovedFromCalculatedTx, the +// legitimate push path, is untouched: it lives on the repository and is called from +// mbpush, never through these handlers. +type MBCostRowChecker interface { + IsMBCostRow(ctx context.Context, costID int64) (bool, error) +} + +// ErrMBCostNotManuallyTransitionable is returned when verify / approve targets a cost row +// belonging to an MB product. +var ErrMBCostNotManuallyTransitionable = errors.New( + "MB cost results are verified and approved by MB Push to Head, not from the cost results screen") + +// rejectMBCostRow fails a manual status transition whose target row is MB-typed. A nil +// checker disables the guard (tests, and any wiring that omits it) — behavior then +// matches the pre-guard code exactly. A checker error is propagated rather than +// swallowed into "not MB", so a database blip cannot let an MB transition through. +func rejectMBCostRow(ctx context.Context, guard MBCostRowChecker, costID int64) error { + if guard == nil { + return nil + } + isMB, err := guard.IsMBCostRow(ctx, costID) + if err != nil { + return fmt.Errorf("check MB cost row: %w", err) + } + if isMB { + return fmt.Errorf("cost %d: %w", costID, ErrMBCostNotManuallyTransitionable) + } + return nil +} diff --git a/services/finance/internal/application/costcalc/mb_cost_row_guard_test.go b/services/finance/internal/application/costcalc/mb_cost_row_guard_test.go new file mode 100644 index 00000000..c092703b --- /dev/null +++ b/services/finance/internal/application/costcalc/mb_cost_row_guard_test.go @@ -0,0 +1,137 @@ +package costcalc + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/require" +) + +// ============================================================================= +// Unit tests for the MB guard on the manual verify / approve RPCs. +// +// MB cost rows go CALCULATED -> APPROVED exclusively inside MB Push-to-Head's +// transaction (CostResultRepository.MarkApprovedFromCalculatedTx), which requires the +// source row to be exactly CALCULATED. A user hand-verifying an MB row through the +// generic Cost Results screen flips it to VERIFIED and makes the next push silently +// SKIP that MB. These tests lock the rejection, and equally lock that a non-MB row is +// untouched. +// +// Service is left empty on the reject paths on purpose: reaching resultRepo would +// nil-panic, which is the assertion that no status was written. +// ============================================================================= + +// stubCostRowChecker answers IsMBCostRow from fixed values and records the ask. +type stubCostRowChecker struct { + isMB bool + err error + askedCost int64 + callsCount int +} + +func (s *stubCostRowChecker) IsMBCostRow(_ context.Context, costID int64) (bool, error) { + s.callsCount++ + s.askedCost = costID + return s.isMB, s.err +} + +func TestVerifyCost_MBRow_Rejected(t *testing.T) { + t.Parallel() + guard := &stubCostRowChecker{isMB: true} + h := NewVerifyCostHandler(&Service{}, WithVerifyMBGuard(guard)) + + err := h.Handle(context.Background(), VerifyCostCommand{CostID: 9001, Actor: "test"}) + + require.Error(t, err) + require.True(t, errors.Is(err, ErrMBCostNotManuallyTransitionable)) + require.Equal(t, int64(9001), guard.askedCost) +} + +func TestApproveCost_MBRow_Rejected(t *testing.T) { + t.Parallel() + guard := &stubCostRowChecker{isMB: true} + h := NewApproveCostHandler(&Service{}, WithApproveMBGuard(guard)) + + err := h.Handle(context.Background(), ApproveCostCommand{CostID: 9002, Actor: "test"}) + + require.Error(t, err) + require.True(t, errors.Is(err, ErrMBCostNotManuallyTransitionable)) + require.Equal(t, int64(9002), guard.askedCost) +} + +// A non-MB row must pass the guard untouched. The handler then proceeds and panics on +// the nil resultRepo — proving the guard did NOT short-circuit a legitimate yarn/POY/ACY +// verification, which is the constraint that matters most here. +func TestVerifyCost_NonMBRow_PassesGuard(t *testing.T) { + t.Parallel() + guard := &stubCostRowChecker{isMB: false} + h := NewVerifyCostHandler(&Service{}, WithVerifyMBGuard(guard)) + + require.Panics(t, func() { + _ = h.Handle(context.Background(), VerifyCostCommand{CostID: 55, Actor: "test"}) + }) + require.Equal(t, 1, guard.callsCount) +} + +func TestApproveCost_NonMBRow_PassesGuard(t *testing.T) { + t.Parallel() + guard := &stubCostRowChecker{isMB: false} + h := NewApproveCostHandler(&Service{}, WithApproveMBGuard(guard)) + + require.Panics(t, func() { + _ = h.Handle(context.Background(), ApproveCostCommand{CostID: 56, Actor: "test"}) + }) + require.Equal(t, 1, guard.callsCount) +} + +// Without the option the guard is disabled entirely and existing call sites keep +// working unchanged. +func TestVerifyApprove_NilGuard_NotConsulted(t *testing.T) { + t.Parallel() + vh := NewVerifyCostHandler(&Service{}) + ah := NewApproveCostHandler(&Service{}) + + require.Panics(t, func() { + _ = vh.Handle(context.Background(), VerifyCostCommand{CostID: 1, Actor: "a"}) + }) + require.Panics(t, func() { + _ = ah.Handle(context.Background(), ApproveCostCommand{CostID: 1, Actor: "a"}) + }) +} + +// Input validation still runs before the guard: a bad command must not spend a query. +func TestVerifyCost_InvalidInput_GuardNotConsulted(t *testing.T) { + t.Parallel() + guard := &stubCostRowChecker{isMB: true} + h := NewVerifyCostHandler(&Service{}, WithVerifyMBGuard(guard)) + + require.Error(t, h.Handle(context.Background(), VerifyCostCommand{CostID: 0, Actor: "test"})) + require.Error(t, h.Handle(context.Background(), VerifyCostCommand{CostID: 5, Actor: ""})) + require.Zero(t, guard.callsCount) +} + +// A checker failure must not degrade into "not MB": an MB row would otherwise slip +// through on a transient database error. +func TestVerifyCost_GuardError_Propagates(t *testing.T) { + t.Parallel() + sentinel := errors.New("db down") + h := NewVerifyCostHandler(&Service{}, WithVerifyMBGuard(&stubCostRowChecker{err: sentinel})) + + err := h.Handle(context.Background(), VerifyCostCommand{CostID: 7, Actor: "test"}) + + require.Error(t, err) + require.True(t, errors.Is(err, sentinel)) + require.False(t, errors.Is(err, ErrMBCostNotManuallyTransitionable)) +} + +func TestApproveCost_GuardError_Propagates(t *testing.T) { + t.Parallel() + sentinel := errors.New("db down") + h := NewApproveCostHandler(&Service{}, WithApproveMBGuard(&stubCostRowChecker{err: sentinel})) + + err := h.Handle(context.Background(), ApproveCostCommand{CostID: 7, Actor: "test"}) + + require.Error(t, err) + require.True(t, errors.Is(err, sentinel)) +} diff --git a/services/finance/internal/application/costcalc/process_chunk.go b/services/finance/internal/application/costcalc/process_chunk.go index 43b2c744..6cd7df5e 100644 --- a/services/finance/internal/application/costcalc/process_chunk.go +++ b/services/finance/internal/application/costcalc/process_chunk.go @@ -29,6 +29,9 @@ const ( blockReasonMissingUpstream = "MISSING_UPSTREAM_COST" blockReasonMissingMBCost = "MISSING_MB_COST" blockReasonFormulaError = "FORMULA_ERROR" + // blockReasonMBOwnedByBatch marks an MB product that reached a generic calc chunk as a + // dependency node rather than as a job target. See computeOne. + blockReasonMBOwnedByBatch = "MB_OWNED_BY_MB_BATCH" ) // ProcessChunkInput is the slice of work passed into ProcessChunk. @@ -123,6 +126,9 @@ type loadedBundle struct { // POY product in the chunk. Its Period may be older than the requested period // (deliberate carry-forward) and is recorded per product in the calculation log. spinPool SpinPool + // mbProducts is the set of this chunk's product ids that are MB-typed. Empty when + // no MB guard is wired. See computeOne for why they are skipped, not computed. + mbProducts map[int64]bool } func (s *Service) bulkLoad(ctx context.Context, in ProcessChunkInput) (*loadedBundle, error) { @@ -151,6 +157,11 @@ func (s *Service) bulkLoad(ctx context.Context, in ProcessChunkInput) (*loadedBu return nil, fmt.Errorf("load upstream costs: %w", err) } + mbProducts, err := s.loadMBProductSet(ctx, in.Products) + if err != nil { + return nil, err + } + sellingSnaps, snapErr := s.loader.LoadSellingSnapshots(ctx, in.Products, in.Period) if snapErr != nil { // Non-fatal: proceed with empty snapshots so marketing_result() returns 0 @@ -187,9 +198,23 @@ func (s *Service) bulkLoad(ctx context.Context, in ProcessChunkInput) (*loadedBu upstreamCosts: upstreamCosts, sellingSnapshots: sellingSnaps, spinPool: spinPool, + mbProducts: mbProducts, }, nil } +// loadMBProductSet resolves which of the chunk's products are MB-typed. A nil guard +// (tests, or wiring that omits it) yields an empty set, i.e. the pre-guard behavior. +func (s *Service) loadMBProductSet(ctx context.Context, products []int64) (map[int64]bool, error) { + if s.mbProductGuard == nil { + return map[int64]bool{}, nil + } + set, err := s.mbProductGuard.MBProductIDs(ctx, products) + if err != nil { + return nil, fmt.Errorf("load MB product set: %w", err) + } + return set, nil +} + type productOutcome int const ( @@ -201,6 +226,30 @@ const ( // computeOne runs the full per-product pipeline: gate on route presence, // compute, persist, mark job_product. func (s *Service) computeOne(ctx context.Context, in ProcessChunkInput, pid int64, loaded *loadedBundle) productOutcome { + // MB products are written exclusively by MB Batch (application/mbbatch), which owns + // all three calc types and whose output MB Push-to-Head consumes. The orchestrator + // already keeps MB out of every job SEED set, but loadProductRMEdges deliberately does + // NOT filter MB during dependency resolution — a non-MB product that referenced an MB + // as a PRODUCT-type RM would therefore pull that MB into the DAG as a node, and + // persistResult would UpsertWithSupersede its cst_product_cost rows, superseding the + // APPROVED row a push already consumed. Refusing at the WRITE site closes that hole + // without touching dependency resolution. + // + // Nothing is lost by skipping: a parent consuming an MB reads the MB's committed value, + // never a value recomputed here — either through the MB_COST_LOOKUP formula path off + // cst_mb_cost (compute.go, FormulaTypeMBCostLookup) or, for a nested-MB PRODUCT-type RM, + // through LoadUpstreamCosts reading the already-committed cst_product_cost row. Both + // read committed state, so this skip cannot change any parent's number. + if loaded.mbProducts[pid] { + if e := s.productRepo.MarkBlocked(ctx, in.JobID, pid, blockReasonMBOwnedByBatch, nil); e != nil { + _ = e + } + s.emitProductBlocked(ctx, in, pid, blockReasonMBOwnedByBatch, + errors.New("MB product is calculated by MB Batch (MB Push to Head), not by the calc engine")) + metrics.ProductsTotal.WithLabelValues(productStatusBlocked, blockReasonMBOwnedByBatch).Inc() + return productOutcomeBlocked + } + route, ok := loaded.routes[pid] if !ok || route == nil || route.Head == nil { if e := s.productRepo.MarkBlocked(ctx, in.JobID, pid, blockReasonMissingRoute, nil); e != nil { diff --git a/services/finance/internal/application/costcalc/process_chunk_mb_guard_test.go b/services/finance/internal/application/costcalc/process_chunk_mb_guard_test.go new file mode 100644 index 00000000..9b48c94a --- /dev/null +++ b/services/finance/internal/application/costcalc/process_chunk_mb_guard_test.go @@ -0,0 +1,231 @@ +package costcalc + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/require" + + "github.com/mutugading/goapps-backend/services/finance/internal/application/costcalc/evaluator" + costcalcdom "github.com/mutugading/goapps-backend/services/finance/internal/domain/costcalc" + "github.com/mutugading/goapps-backend/services/finance/internal/domain/costroute" +) + +// ============================================================================= +// Unit tests for the persist-site MB guard in ProcessChunk / computeOne. +// +// The orchestrator's loadProductRMEdges deliberately does NOT filter MB, so the MB +// exclusion constrains what a job TARGETS, not how dependencies are RESOLVED. That +// leaves one hole: a non-MB product with a PRODUCT-type RM edge to an MB would pull +// the MB into the DAG as a node, and persistResult would UpsertWithSupersede its +// cst_product_cost rows — re-creating the very bug the exclusion closed. The guard +// refuses at the WRITE site instead. +// +// The suite locks BOTH halves: MB is skipped, and every non-MB product in the same +// chunk is still computed and persisted. The second half is the regression guard for +// the standing "do not disturb yarn/POY/ACY" constraint. +// ============================================================================= + +// mbGuardLoader answers every bulkLoad dependency with a usable minimum: a route per +// requested product (so no product is blocked as MISSING_ROUTE), no formulas, and a +// non-empty spin pool (bulkLoad hard-blocks on an empty one). +type mbGuardLoader struct { + ProductLoader + products []int64 +} + +func (f *mbGuardLoader) LoadRoutesByProducts(_ context.Context, ids []int64) (map[int64]*costroute.Graph, error) { + out := map[int64]*costroute.Graph{} + for _, id := range ids { + out[id] = &costroute.Graph{Head: &costroute.Head{HeadID: id * 10}} + } + return out, nil +} + +func (f *mbGuardLoader) LoadCAPP(_ context.Context, _ []int64) (map[int64]map[string]float64, error) { + return map[int64]map[string]float64{}, nil +} + +func (f *mbGuardLoader) LoadFormulas(_ context.Context, _ []int64) (map[int64][]Formula, error) { + return map[int64][]Formula{}, nil +} + +func (f *mbGuardLoader) LoadRMCosts(_ context.Context, _ []string, _, _ string) (map[string]float64, error) { + return map[string]float64{}, nil +} + +func (f *mbGuardLoader) LoadUpstreamCosts(_ context.Context, _ []int64, _, _ string) (map[int64]float64, error) { + return map[int64]float64{}, nil +} + +func (f *mbGuardLoader) LoadSellingSnapshots(_ context.Context, _ []int64, _ string) (map[int64]map[string]float64, error) { + return map[int64]map[string]float64{}, nil +} + +func (f *mbGuardLoader) LoadSpinFixedCost(_ context.Context, period string) (SpinPool, error) { + return fullPool(period), nil +} + +// mbSetChecker reports a fixed set of MB product ids. +type mbSetChecker struct { + mb map[int64]bool + err error + calls int + asked []int64 +} + +func (c *mbSetChecker) MBProductIDs(_ context.Context, ids []int64) (map[int64]bool, error) { + c.calls++ + c.asked = ids + return c.mb, c.err +} + +// recordingProductRepo captures the per-product terminal transitions. +type recordingProductRepo struct { + costcalcdom.JobProductRepository + success map[int64]bool + blocked map[int64]string + failed map[int64]string +} + +func newRecordingProductRepo() *recordingProductRepo { + return &recordingProductRepo{ + success: map[int64]bool{}, + blocked: map[int64]string{}, + failed: map[int64]string{}, + } +} + +func (r *recordingProductRepo) MarkSuccess(_ context.Context, _, pid, _ int64, _ int, _ []byte) error { + r.success[pid] = true + return nil +} + +func (r *recordingProductRepo) MarkBlocked(_ context.Context, _, pid int64, reason string, _ []byte) error { + r.blocked[pid] = reason + return nil +} + +func (r *recordingProductRepo) MarkFailed(_ context.Context, _, pid int64, msg string, _ []byte) error { + r.failed[pid] = msg + return nil +} + +// recordingResultRepo captures which products were persisted into cst_product_cost. +type recordingResultRepo struct { + costcalcdom.ResultRepository + upserted []int64 +} + +func (r *recordingResultRepo) UpsertWithSupersede(_ context.Context, res *costcalcdom.Result) (int64, int, float64, int64, error) { + r.upserted = append(r.upserted, res.ProductSysID()) + return 1, 0, 0, 0, nil +} + +// nopChunkRepo satisfies the chunk transitions ProcessChunk performs when ChunkID != 0. +// The suite passes ChunkID 0, so these are never called; the embedded nil interface +// would panic loudly if that ever changed. +type nopChunkRepo struct{ costcalcdom.ChunkRepository } + +func mbGuardService(guard MBProductSetChecker, prodRepo *recordingProductRepo, resRepo *recordingResultRepo) *Service { + return NewService( + nil, &nopChunkRepo{}, prodRepo, resRepo, nil, + &mbGuardLoader{}, evaluator.NewCache(), nil, nil, + WithMBProductGuard(guard), + ) +} + +func mbGuardInput(products []int64) ProcessChunkInput { + return ProcessChunkInput{ + JobID: 42, + ChunkID: 0, + Period: "202607", + CalcType: costcalcdom.CalcTypeActual, + Products: products, + Actor: "mb-persist-guard-test", + } +} + +// The core assertion: an MB dependency node in a mixed chunk is BLOCKED, never written, +// while every non-MB product in the same chunk computes and persists exactly as before. +func TestProcessChunk_MBDependencyNode_NotPersisted(t *testing.T) { + t.Parallel() + prodRepo := newRecordingProductRepo() + resRepo := &recordingResultRepo{} + guard := &mbSetChecker{mb: map[int64]bool{777: true}} + svc := mbGuardService(guard, prodRepo, resRepo) + + out, err := svc.ProcessChunk(context.Background(), mbGuardInput([]int64{101, 777, 202})) + require.NoError(t, err) + + // MB skipped, not written. + require.NotContains(t, resRepo.upserted, int64(777)) + require.Equal(t, blockReasonMBOwnedByBatch, prodRepo.blocked[777]) + require.False(t, prodRepo.success[777]) + + // Non-MB products in the SAME chunk are untouched by the guard. + require.Contains(t, resRepo.upserted, int64(101)) + require.Contains(t, resRepo.upserted, int64(202)) + require.True(t, prodRepo.success[101]) + require.True(t, prodRepo.success[202]) + require.NotContains(t, prodRepo.blocked, int64(101)) + require.NotContains(t, prodRepo.blocked, int64(202)) + + require.Equal(t, 2, out.Success) + require.Equal(t, 1, out.Blocked) + require.Equal(t, 0, out.Failed) +} + +// A chunk with no MB at all must behave byte-for-byte as before: every product persists. +// This is the direct regression guard for yarn / POY / ACY. +func TestProcessChunk_AllNonMB_EveryProductPersisted(t *testing.T) { + t.Parallel() + prodRepo := newRecordingProductRepo() + resRepo := &recordingResultRepo{} + guard := &mbSetChecker{mb: map[int64]bool{}} + svc := mbGuardService(guard, prodRepo, resRepo) + + products := []int64{11, 22, 33, 44} + out, err := svc.ProcessChunk(context.Background(), mbGuardInput(products)) + require.NoError(t, err) + + require.ElementsMatch(t, products, resRepo.upserted) + require.Empty(t, prodRepo.blocked) + require.Empty(t, prodRepo.failed) + require.Equal(t, len(products), out.Success) + require.Equal(t, 1, guard.calls, "the guard must cost one query per chunk, not one per product") + require.ElementsMatch(t, products, guard.asked) +} + +// Without WithMBProductGuard the guard is disabled and nothing is skipped — existing +// call sites (tests, any wiring that omits it) keep their exact behavior. +func TestProcessChunk_NilGuard_NothingSkipped(t *testing.T) { + t.Parallel() + prodRepo := newRecordingProductRepo() + resRepo := &recordingResultRepo{} + svc := NewService(nil, &nopChunkRepo{}, prodRepo, resRepo, nil, &mbGuardLoader{}, evaluator.NewCache(), nil, nil) + + out, err := svc.ProcessChunk(context.Background(), mbGuardInput([]int64{5, 6})) + require.NoError(t, err) + + require.ElementsMatch(t, []int64{5, 6}, resRepo.upserted) + require.Equal(t, 2, out.Success) + require.Empty(t, prodRepo.blocked) +} + +// A guard failure must abort the chunk rather than degrade into "nothing is MB", which +// would silently re-open the write hole. +func TestProcessChunk_GuardError_AbortsChunk(t *testing.T) { + t.Parallel() + sentinel := errors.New("db down") + prodRepo := newRecordingProductRepo() + resRepo := &recordingResultRepo{} + svc := mbGuardService(&mbSetChecker{err: sentinel}, prodRepo, resRepo) + + _, err := svc.ProcessChunk(context.Background(), mbGuardInput([]int64{101})) + + require.Error(t, err) + require.True(t, errors.Is(err, sentinel)) + require.Empty(t, resRepo.upserted, "no cost row may be written when MB membership is unknown") +} diff --git a/services/finance/internal/application/costcalc/service.go b/services/finance/internal/application/costcalc/service.go index d5c9b05b..674bcf5f 100644 --- a/services/finance/internal/application/costcalc/service.go +++ b/services/finance/internal/application/costcalc/service.go @@ -20,6 +20,23 @@ type Service struct { cache *evaluator.Cache auditEmitter AuditEmitter jobTriggerPub JobTriggerPublisher + + mbProductGuard MBProductSetChecker +} + +// MBProductSetChecker resolves which of a chunk's products are MB-typed, so ProcessChunk +// can refuse to persist a cst_product_cost row for an MB. See computeOne for the full +// rationale. nil disables the guard. +type MBProductSetChecker interface { + MBProductIDs(ctx context.Context, productSysIDs []int64) (map[int64]bool, error) +} + +// ServiceOption customizes the service at construction. +type ServiceOption func(*Service) + +// WithMBProductGuard installs the persist-site MB guard. +func WithMBProductGuard(c MBProductSetChecker) ServiceOption { + return func(s *Service) { s.mbProductGuard = c } } // JobTriggerPublisher signals the orchestrator (via RMQ) to plan + execute a @@ -65,8 +82,9 @@ func NewService( cache *evaluator.Cache, auditEmitter AuditEmitter, jobTriggerPub JobTriggerPublisher, + opts ...ServiceOption, ) *Service { - return &Service{ + s := &Service{ jobRepo: jobRepo, chunkRepo: chunkRepo, productRepo: productRepo, @@ -77,6 +95,10 @@ func NewService( auditEmitter: auditEmitter, jobTriggerPub: jobTriggerPub, } + for _, opt := range opts { + opt(s) + } + return s } // emitAudit is a best-effort fire-and-forget helper: nil emitter means skip, diff --git a/services/finance/internal/application/costcalc/verify_cost_handler.go b/services/finance/internal/application/costcalc/verify_cost_handler.go index 170c457c..13f74f94 100644 --- a/services/finance/internal/application/costcalc/verify_cost_handler.go +++ b/services/finance/internal/application/costcalc/verify_cost_handler.go @@ -17,12 +17,26 @@ type VerifyCostCommand struct { // VerifyCostHandler transitions a CALCULATED result to VERIFIED. type VerifyCostHandler struct { - svc *Service + svc *Service + mbGuard MBCostRowChecker +} + +// VerifyOption customizes the handler at construction. +type VerifyOption func(*VerifyCostHandler) + +// WithVerifyMBGuard installs the MB rejection check. Omitting it leaves the check +// disabled; tests omit it. +func WithVerifyMBGuard(c MBCostRowChecker) VerifyOption { + return func(h *VerifyCostHandler) { h.mbGuard = c } } // NewVerifyCostHandler constructs the handler. -func NewVerifyCostHandler(svc *Service) *VerifyCostHandler { - return &VerifyCostHandler{svc: svc} +func NewVerifyCostHandler(svc *Service, opts ...VerifyOption) *VerifyCostHandler { + h := &VerifyCostHandler{svc: svc} + for _, opt := range opts { + opt(h) + } + return h } // Handle executes the verification. @@ -33,6 +47,9 @@ func (h *VerifyCostHandler) Handle(ctx context.Context, cmd VerifyCostCommand) e if cmd.Actor == "" { return errors.New(errMsgActorRequired) } + if err := rejectMBCostRow(ctx, h.mbGuard, cmd.CostID); err != nil { + return err + } if err := h.svc.resultRepo.MarkVerified(ctx, cmd.CostID, cmd.Actor); err != nil { return fmt.Errorf("mark verified: %w", err) } diff --git a/services/finance/internal/application/mbpush/port.go b/services/finance/internal/application/mbpush/port.go index f942a92f..9f205697 100644 --- a/services/finance/internal/application/mbpush/port.go +++ b/services/finance/internal/application/mbpush/port.go @@ -35,6 +35,14 @@ type CostReader interface { MarkApprovedFromCalculatedTx(ctx context.Context, tx *sql.Tx, costID int64, by string) error } +// StalePushReader reports which MB Heads were already pushed for a period but whose pushed cost +// has since been superseded by a later MB Batch run. Read-only: it informs Preview's labeling +// and never gates a write. +type StalePushReader interface { + // ListStalePushedMBHIDs returns the mbh_ids in a stale-push state for period. + ListStalePushedMBHIDs(ctx context.Context, period string) ([]string, error) +} + // MBCostWriter upserts the active-cost cache row inside a caller-supplied transaction. type MBCostWriter interface { Upsert(ctx context.Context, tx *sql.Tx, mbhID, period, costType, costValue string, sourceCpcID int64, pushedBy string) error diff --git a/services/finance/internal/application/mbpush/preview_handler.go b/services/finance/internal/application/mbpush/preview_handler.go index 8a450718..fa4eee09 100644 --- a/services/finance/internal/application/mbpush/preview_handler.go +++ b/services/finance/internal/application/mbpush/preview_handler.go @@ -21,6 +21,10 @@ type PushableMBHead struct { HasActual bool HasSelling bool HasForecast bool + // NeedsRepush marks a head already pushed for the period whose pushed cost went stale after a + // later MB Batch run. Informational only — the head stays pushable, since re-pushing is the + // remedy. Never auto-acted on. + NeedsRepush bool } // SkippedMBHead is an MB Head excluded from the push, with the reason(s) why. @@ -31,40 +35,78 @@ type SkippedMBHead struct { Reason string } +// PreviewResult is the full outcome of a push preview: the pushable set, the skipped set, and +// the count of pushable heads that are pushable *because* their previous push went stale. +type PreviewResult struct { + Pushable []PushableMBHead + Skipped []SkippedMBHead + NeedsRepushCount int32 +} + // PreviewHandler computes which VALIDATED MB Heads are ready for a push-to-head execution. type PreviewHandler struct { - mbHeadReader MBHeadReader - costReader CostReader + mbHeadReader MBHeadReader + costReader CostReader + stalePushReader StalePushReader } // NewPreviewHandler constructs a PreviewHandler. -func NewPreviewHandler(mbHeadReader MBHeadReader, costReader CostReader) *PreviewHandler { - return &PreviewHandler{mbHeadReader: mbHeadReader, costReader: costReader} +func NewPreviewHandler(mbHeadReader MBHeadReader, costReader CostReader, stalePushReader StalePushReader) *PreviewHandler { + return &PreviewHandler{mbHeadReader: mbHeadReader, costReader: costReader, stalePushReader: stalePushReader} } // Preview lists VALIDATED MB Heads split into pushable (all 3 cost types CALCULATED) and skipped -// (with the reason(s) why), per PR-02. -func (h *PreviewHandler) Preview(ctx context.Context, period string) ([]PushableMBHead, []SkippedMBHead, error) { +// (with the reason(s) why), per PR-02. Pushable heads whose already-pushed cost has gone stale +// after a later MB Batch run are additionally flagged NeedsRepush — a label only, it neither +// moves a head between buckets nor pushes anything. +func (h *PreviewHandler) Preview(ctx context.Context, period string) (*PreviewResult, error) { candidates, err := h.mbHeadReader.ListValidated(ctx) if err != nil { - return nil, nil, fmt.Errorf("list validated mb heads: %w", err) + return nil, fmt.Errorf("list validated mb heads: %w", err) + } + stale, err := h.staleSet(ctx, period) + if err != nil { + return nil, err } - pushable := make([]PushableMBHead, 0, len(candidates)) - skipped := make([]SkippedMBHead, 0, len(candidates)) + result := &PreviewResult{ + Pushable: make([]PushableMBHead, 0, len(candidates)), + Skipped: make([]SkippedMBHead, 0, len(candidates)), + } for _, c := range candidates { if c.CostProductID == 0 { - skipped = append(skipped, SkippedMBHead{MBHID: c.MBHID, Code: c.Code, Name: c.Name, Reason: skipReasonNoCostProduct}) + result.Skipped = append(result.Skipped, SkippedMBHead{MBHID: c.MBHID, Code: c.Code, Name: c.Name, Reason: skipReasonNoCostProduct}) continue } p, reasons := h.checkCostTypes(ctx, c, period) if len(reasons) > 0 { - skipped = append(skipped, SkippedMBHead{MBHID: c.MBHID, Code: c.Code, Name: c.Name, Reason: strings.Join(reasons, ", ")}) + result.Skipped = append(result.Skipped, SkippedMBHead{MBHID: c.MBHID, Code: c.Code, Name: c.Name, Reason: strings.Join(reasons, ", ")}) continue } - pushable = append(pushable, p) + if _, ok := stale[c.MBHID]; ok { + p.NeedsRepush = true + result.NeedsRepushCount++ + } + result.Pushable = append(result.Pushable, p) + } + return result, nil +} + +// staleSet resolves the stale-push mbh_id set for period. A nil reader yields an empty set so +// the flag degrades to "never set" rather than breaking the preview. +func (h *PreviewHandler) staleSet(ctx context.Context, period string) (map[string]struct{}, error) { + if h.stalePushReader == nil { + return map[string]struct{}{}, nil + } + ids, err := h.stalePushReader.ListStalePushedMBHIDs(ctx, period) + if err != nil { + return nil, fmt.Errorf("list stale pushed mb heads: %w", err) + } + set := make(map[string]struct{}, len(ids)) + for _, id := range ids { + set[id] = struct{}{} } - return pushable, skipped, nil + return set, nil } func (h *PreviewHandler) checkCostTypes(ctx context.Context, c MBHeadCandidate, period string) (PushableMBHead, []string) { diff --git a/services/finance/internal/application/mbpush/preview_handler_test.go b/services/finance/internal/application/mbpush/preview_handler_test.go new file mode 100644 index 00000000..43f75081 --- /dev/null +++ b/services/finance/internal/application/mbpush/preview_handler_test.go @@ -0,0 +1,203 @@ +package mbpush + +import ( + "context" + "database/sql" + "errors" + "testing" + + "github.com/stretchr/testify/require" +) + +// fakeMBHeadReader returns a fixed candidate set. +type fakeMBHeadReader struct { + candidates []MBHeadCandidate + err error +} + +func (f *fakeMBHeadReader) ListValidated(context.Context) ([]MBHeadCandidate, error) { + return f.candidates, f.err +} + +// fakeCostReader reports which (productSysID, costType) tuples have an active CALCULATED row. +// Only the Preview path (GetActiveCalculated) is exercised; the Tx methods are unused here. +type fakeCostReader struct { + // missing lists the cost types with NO calculated row, keyed by product sys id. + missing map[int64]map[string]bool +} + +func (f *fakeCostReader) GetActiveCalculated(_ context.Context, productSysID int64, _, calcType string) (int64, string, bool, error) { + if f.missing[productSysID][calcType] { + return 0, "", false, nil + } + return productSysID, "1.000000", true, nil +} + +func (f *fakeCostReader) GetActiveCalculatedTx(context.Context, *sql.Tx, int64, string, string) (int64, string, bool, error) { + return 0, "", false, nil +} + +func (f *fakeCostReader) MarkApprovedFromCalculatedTx(context.Context, *sql.Tx, int64, string) error { + return nil +} + +// fakeStalePushReader returns a fixed stale-push id set and records the period it was asked for. +type fakeStalePushReader struct { + ids []string + err error + gotPeriod string + callCount int +} + +func (f *fakeStalePushReader) ListStalePushedMBHIDs(_ context.Context, period string) ([]string, error) { + f.callCount++ + f.gotPeriod = period + if f.err != nil { + return nil, f.err + } + return f.ids, nil +} + +const testPeriod = "202608" + +func newPreviewHandler(candidates []MBHeadCandidate, missing map[int64]map[string]bool, stale StalePushReader) *PreviewHandler { + return NewPreviewHandler( + &fakeMBHeadReader{candidates: candidates}, + &fakeCostReader{missing: missing}, + stale, + ) +} + +// The core of this feature: a head that is fully calculated AND in the stale-push set stays in +// the pushable bucket (re-pushing is the remedy) but carries the NeedsRepush label, and the +// count reflects it. +func TestPreview_StalePushedHeadStaysPushableAndIsFlagged(t *testing.T) { + h := newPreviewHandler( + []MBHeadCandidate{ + {MBHID: "mb-stale", Code: "MB-1", Name: "Stale", CostProductID: 101}, + {MBHID: "mb-fresh", Code: "MB-2", Name: "Fresh", CostProductID: 102}, + }, + nil, + &fakeStalePushReader{ids: []string{"mb-stale"}}, + ) + + result, err := h.Preview(context.Background(), testPeriod) + require.NoError(t, err) + + require.Len(t, result.Pushable, 2, "a stale push must not be demoted out of the pushable bucket") + require.Empty(t, result.Skipped) + require.Equal(t, int32(1), result.NeedsRepushCount) + + byID := map[string]PushableMBHead{} + for _, p := range result.Pushable { + byID[p.MBHID] = p + } + require.True(t, byID["mb-stale"].NeedsRepush) + require.False(t, byID["mb-fresh"].NeedsRepush) +} + +// A stale-push row for a head that cannot be pushed at all (missing cost types) must not leak +// into the count — the count only ever describes heads present in the pushable bucket. +func TestPreview_StaleFlagNotCountedForSkippedHead(t *testing.T) { + h := newPreviewHandler( + []MBHeadCandidate{{MBHID: "mb-stale", Code: "MB-1", Name: "Stale", CostProductID: 101}}, + map[int64]map[string]bool{101: {"SELLING": true}}, + &fakeStalePushReader{ids: []string{"mb-stale"}}, + ) + + result, err := h.Preview(context.Background(), testPeriod) + require.NoError(t, err) + + require.Empty(t, result.Pushable) + require.Len(t, result.Skipped, 1) + require.Equal(t, skipReasonMissingSelling, result.Skipped[0].Reason) + require.Zero(t, result.NeedsRepushCount) +} + +// A head with no linked cost product is skipped before any cost lookup, and the stale set — +// which is keyed by mbh_id and could still contain it — must not flag it. +func TestPreview_NoCostProductSkippedRegardlessOfStaleSet(t *testing.T) { + h := newPreviewHandler( + []MBHeadCandidate{{MBHID: "mb-nolink", Code: "MB-3", Name: "No link", CostProductID: 0}}, + nil, + &fakeStalePushReader{ids: []string{"mb-nolink"}}, + ) + + result, err := h.Preview(context.Background(), testPeriod) + require.NoError(t, err) + + require.Empty(t, result.Pushable) + require.Len(t, result.Skipped, 1) + require.Equal(t, skipReasonNoCostProduct, result.Skipped[0].Reason) + require.Zero(t, result.NeedsRepushCount) +} + +// With nothing stale, the preview must behave exactly as it did before this feature. +func TestPreview_NoStaleRowsLeavesEveryHeadUnflagged(t *testing.T) { + stale := &fakeStalePushReader{} + h := newPreviewHandler( + []MBHeadCandidate{ + {MBHID: "mb-a", Code: "MB-A", Name: "A", CostProductID: 1}, + {MBHID: "mb-b", Code: "MB-B", Name: "B", CostProductID: 2}, + }, + nil, + stale, + ) + + result, err := h.Preview(context.Background(), testPeriod) + require.NoError(t, err) + + require.Len(t, result.Pushable, 2) + require.Zero(t, result.NeedsRepushCount) + for _, p := range result.Pushable { + require.False(t, p.NeedsRepush) + } + require.Equal(t, testPeriod, stale.gotPeriod, "the stale lookup must be scoped to the previewed period") + require.Equal(t, 1, stale.callCount, "the stale set is resolved once per preview, not per candidate") +} + +// The stale lookup is informational, but it is still part of the response contract: a failure +// surfaces as an error rather than silently under-reporting. +func TestPreview_StaleReaderErrorPropagates(t *testing.T) { + h := newPreviewHandler( + []MBHeadCandidate{{MBHID: "mb-a", Code: "MB-A", Name: "A", CostProductID: 1}}, + nil, + &fakeStalePushReader{err: errors.New("boom")}, + ) + + _, err := h.Preview(context.Background(), testPeriod) + require.Error(t, err) + require.Contains(t, err.Error(), "list stale pushed mb heads") +} + +// A nil reader must degrade to "never flagged" rather than panicking, so an incompletely wired +// preview still returns its pushable/skipped buckets. +func TestPreview_NilStaleReaderDegradesToUnflagged(t *testing.T) { + h := newPreviewHandler( + []MBHeadCandidate{{MBHID: "mb-a", Code: "MB-A", Name: "A", CostProductID: 1}}, + nil, + nil, + ) + + result, err := h.Preview(context.Background(), testPeriod) + require.NoError(t, err) + require.Len(t, result.Pushable, 1) + require.False(t, result.Pushable[0].NeedsRepush) + require.Zero(t, result.NeedsRepushCount) +} + +// Every missing cost type must still be reported, unchanged by the new classification. +func TestPreview_AllCostTypesMissingReportsEveryReason(t *testing.T) { + h := newPreviewHandler( + []MBHeadCandidate{{MBHID: "mb-a", Code: "MB-A", Name: "A", CostProductID: 1}}, + map[int64]map[string]bool{1: {"ACTUAL": true, "SELLING": true, "FORECAST": true}}, + &fakeStalePushReader{}, + ) + + result, err := h.Preview(context.Background(), testPeriod) + require.NoError(t, err) + require.Len(t, result.Skipped, 1) + require.Contains(t, result.Skipped[0].Reason, skipReasonMissingActual) + require.Contains(t, result.Skipped[0].Reason, skipReasonMissingSelling) + require.Contains(t, result.Skipped[0].Reason, skipReasonMissingForecast) +} diff --git a/services/finance/internal/delivery/grpc/cost_calc_handler.go b/services/finance/internal/delivery/grpc/cost_calc_handler.go index b5ca1ed0..4fea71f7 100644 --- a/services/finance/internal/delivery/grpc/cost_calc_handler.go +++ b/services/finance/internal/delivery/grpc/cost_calc_handler.go @@ -1239,6 +1239,7 @@ func mappedCostCalcErrToBase(err error) *commonv1.BaseResponse { return ErrorResponse("501", err.Error()) case errors.Is(err, costcalc.ErrProductRequired), errors.Is(err, costcalc.ErrMBNotCalcJobEligible), + errors.Is(err, costcalc.ErrMBCostNotManuallyTransitionable), errors.Is(err, costcalcdom.ErrInvalidPeriod): return ErrorResponse("400", err.Error()) // Configuration gap, not a server fault: Finance has not entered a diff --git a/services/finance/internal/delivery/grpc/mb_push_handler.go b/services/finance/internal/delivery/grpc/mb_push_handler.go index 2ced9472..40bfb34b 100644 --- a/services/finance/internal/delivery/grpc/mb_push_handler.go +++ b/services/finance/internal/delivery/grpc/mb_push_handler.go @@ -42,7 +42,7 @@ func (h *MBPushHandler) PreviewPushToHead(ctx context.Context, req *financev1.Pr return &financev1.PreviewPushToHeadResponse{Base: baseResp}, nil } - pushable, skipped, err := h.previewHandler.Preview(ctx, req.Period) + result, err := h.previewHandler.Preview(ctx, req.Period) if err != nil { RecordMBPushOperation("preview", false) return &financev1.PreviewPushToHeadResponse{Base: domainErrorToBaseResponse(err)}, nil @@ -50,19 +50,20 @@ func (h *MBPushHandler) PreviewPushToHead(ctx context.Context, req *financev1.Pr RecordMBPushOperation("preview", true) - pushableItems := make([]*financev1.PushableMbHead, len(pushable)) - for i, p := range pushable { + pushableItems := make([]*financev1.PushableMbHead, len(result.Pushable)) + for i, p := range result.Pushable { pushableItems[i] = pushableMBHeadToProto(p) } - skippedItems := make([]*financev1.SkippedMbHead, len(skipped)) - for i, s := range skipped { + skippedItems := make([]*financev1.SkippedMbHead, len(result.Skipped)) + for i, s := range result.Skipped { skippedItems[i] = skippedMBHeadToProto(s) } return &financev1.PreviewPushToHeadResponse{ - Base: successResponse("MB push preview retrieved successfully"), - Pushable: pushableItems, - Skipped: skippedItems, + Base: successResponse("MB push preview retrieved successfully"), + Pushable: pushableItems, + Skipped: skippedItems, + NeedsRepushCount: result.NeedsRepushCount, }, nil } @@ -146,6 +147,7 @@ func pushableMBHeadToProto(p appmbpush.PushableMBHead) *financev1.PushableMbHead HasActual: p.HasActual, HasSelling: p.HasSelling, HasForecast: p.HasForecast, + NeedsRepush: p.NeedsRepush, } } diff --git a/services/finance/internal/infrastructure/postgres/cst_mb_cost_repository.go b/services/finance/internal/infrastructure/postgres/cst_mb_cost_repository.go index 8dd438b0..93f73ab5 100644 --- a/services/finance/internal/infrastructure/postgres/cst_mb_cost_repository.go +++ b/services/finance/internal/infrastructure/postgres/cst_mb_cost_repository.go @@ -38,6 +38,51 @@ func (r *CstMBCostRepository) Upsert(ctx context.Context, tx *sql.Tx, mbhID, per return nil } +// ListStalePushedMBHIDs returns the distinct mbh_ids whose already-pushed cost for period has +// gone stale: an active cst_mb_cost row exists, but its source cst_product_cost row is either +// unlinked (mbc_source_cpc_id IS NULL, e.g. the FK's ON DELETE SET NULL fired) or SUPERSEDED, +// while a newer non-superseded row exists for the same product/period/cost type. That happens +// when MB Batch re-runs after a push — the pushed value is stale and silently so, hence this +// read-only visibility query. Never used to gate a write; Preview only labels the head. +func (r *CstMBCostRepository) ListStalePushedMBHIDs(ctx context.Context, period string) ([]string, error) { + const q = ` + SELECT DISTINCT mbc.mbc_mbh_id::text + FROM cst_mb_cost mbc + JOIN mst_mb_head mbh + ON mbh.mbh_id = mbc.mbc_mbh_id + AND mbh.mbh_entry_status = 'VALIDATED' + AND mbh.deleted_at IS NULL + LEFT JOIN cst_product_cost src ON src.cpc_cost_id = mbc.mbc_source_cpc_id + WHERE mbc.mbc_period = $1 + AND mbc.mbc_is_active = TRUE + AND (mbc.mbc_source_cpc_id IS NULL OR src.cpc_status = 'SUPERSEDED') + AND EXISTS ( + SELECT 1 FROM cst_product_cost cur + WHERE cur.cpc_product_sys_id = mbh.mbh_cost_product_id + AND cur.cpc_period = mbc.mbc_period + AND cur.cpc_calculation_type = mbc.mbc_cost_type + AND cur.cpc_status != 'SUPERSEDED' + )` + rows, err := r.db.QueryContext(ctx, q, period) + if err != nil { + return nil, fmt.Errorf("cst_mb_cost_repository: list stale pushed mbh ids: %w", err) + } + defer closeRows(rows) + + var out []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, fmt.Errorf("cst_mb_cost_repository: scan stale pushed mbh id: %w", err) + } + out = append(out, id) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("cst_mb_cost_repository: iterate stale pushed mbh ids: %w", err) + } + return out, nil +} + // LatestByType returns the most recent active cost_value for mbhID + costType, used by // Plan 04's LoadMBCosts calc-engine loader — the sole read path for MB cost consumers. func (r *CstMBCostRepository) LatestByType(ctx context.Context, mbhID, costType string) (string, error) { diff --git a/services/finance/internal/infrastructure/postgres/mb_type_checker.go b/services/finance/internal/infrastructure/postgres/mb_type_checker.go index bfdc0839..7d15227e 100644 --- a/services/finance/internal/infrastructure/postgres/mb_type_checker.go +++ b/services/finance/internal/infrastructure/postgres/mb_type_checker.go @@ -3,6 +3,8 @@ package postgres import ( "context" "fmt" + + "github.com/lib/pq" ) // MBTypeChecker answers MB-typed questions for the calc-job trigger guard. It @@ -39,6 +41,62 @@ func (c *MBTypeChecker) IsMBProduct(ctx context.Context, productSysID int64) (bo return isMB, nil } +// IsMBCostRow reports whether the cst_product_cost row identified by costID belongs to an +// MB-typed product. It implements costcalc.MBCostRowChecker, guarding the manual +// verify/approve RPCs. +// +// A missing cost row reports false so the repository's own ErrCostInvalidStatus / +// not-found handling stays the single source of truth for "no such cost". +func (c *MBTypeChecker) IsMBCostRow(ctx context.Context, costID int64) (bool, error) { + const q = ` + SELECT EXISTS ( + SELECT 1 + FROM cst_product_cost pc + JOIN cost_product_master pm ON pm.cpm_product_sys_id = pc.cpc_product_sys_id + JOIN cost_product_type pt ON pt.cpt_type_id = pm.cpm_product_type_id + WHERE pc.cpc_cost_id = $1 AND pt.cpt_type_code = $2 + )` + var isMB bool + if err := c.db.QueryRowContext(ctx, q, costID, mbCostProductTypeCode).Scan(&isMB); err != nil { + return false, fmt.Errorf("check MB cost row %d: %w", costID, err) + } + return isMB, nil +} + +// MBProductIDs returns the subset of productSysIDs that are MB-typed, as a set. It +// implements costcalc.MBProductSetChecker, which the chunk processor uses to refuse to +// persist a cst_product_cost row for an MB product pulled into a DAG as a dependency. +// +// One query per chunk, not per product. An empty input returns an empty set without +// touching the database. +func (c *MBTypeChecker) MBProductIDs(ctx context.Context, productSysIDs []int64) (map[int64]bool, error) { + out := map[int64]bool{} + if len(productSysIDs) == 0 { + return out, nil + } + const q = ` + SELECT pm.cpm_product_sys_id + FROM cost_product_master pm + JOIN cost_product_type pt ON pt.cpt_type_id = pm.cpm_product_type_id + WHERE pm.cpm_product_sys_id = ANY($1) AND pt.cpt_type_code = $2` + rows, err := c.db.QueryContext(ctx, q, pq.Array(productSysIDs), mbCostProductTypeCode) + if err != nil { + return nil, fmt.Errorf("load MB product ids: %w", err) + } + defer closeRows(rows) + for rows.Next() { + var id int64 + if err := rows.Scan(&id); err != nil { + return nil, fmt.Errorf("scan MB product id: %w", err) + } + out[id] = true + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate MB product ids: %w", err) + } + return out, nil +} + // IsMBProductType reports whether the product type id is the MB type. func (c *MBTypeChecker) IsMBProductType(ctx context.Context, productTypeID int32) (bool, error) { const q = `SELECT EXISTS (SELECT 1 FROM cost_product_type WHERE cpt_type_id = $1 AND cpt_type_code = $2)`