apps/backend/src/modules/costing/workers/cost-recalculation.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { randomUUID } from 'node:crypto';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { CostingConfigurationService } from '../application/costing-configuration.service';
import { StockItemCostService } from '../application/stock-item-cost.service';
import { RecipeCostService } from '../application/recipe-cost.service';
import { ProductMarginService } from '../application/product-margin.service';
import { ModifierCostService } from '../application/modifier-cost.service';
import {
assertCostingEnum,
assertCostingId,
assertCostingSafeInteger,
assertCostingScope,
assertCostingText,
} from '../application/costing-input-validation';
@Injectable()
export class CostRecalculationWorker {
constructor(
private readonly database: DatabaseService,
private readonly config: CostingConfigurationService,
private readonly stockCosts: StockItemCostService,
private readonly recipeCosts: RecipeCostService,
private readonly margins: ProductMarginService,
private readonly modifierCosts: ModifierCostService,
) {}
async enqueue(input: {
organizationId: string;
locationId: string;
scopeType: 'all'|'stock_item'|'ingredient'|'recipe'|'product';
scopeId?: string;
reason: string;
}) {
assertCostingScope(input, 'costing.recalculation_input_invalid');
assertCostingEnum(input.scopeType, ['all','stock_item','ingredient','recipe','product'], 'costing.recalculation_input_invalid');
if (input.scopeId !== undefined) assertCostingId(input.scopeId, 'costing.recalculation_input_invalid');
assertCostingText(input.reason, 1000, 'costing.recalculation_input_invalid');
return this.database.transaction(async (tx) => {
// Coalescing is a read/create pair. Serialize only the identical scope key so
// concurrent cost triggers cannot create duplicate pending jobs.
await tx.query(
`select pg_advisory_xact_lock(hashtextextended($1,0))`,
[`cost-recalculation:${input.organizationId}:${input.locationId}:${input.scopeType}:${input.scopeId ?? 'null'}`],
);
const existing = await tx.query<{ id: string }>(
`select id
from cost_recalculation_jobs
where organization_id=$1
and location_id=$2
and scope_type=$3
and scope_id is not distinct from $4
and status='pending'
limit 1`,
[input.organizationId,input.locationId,input.scopeType,input.scopeId ?? null],
);
if (existing[0]) return { id: existing[0].id, coalesced: true };
const id = randomUUID();
await tx.query(
`insert into cost_recalculation_jobs(
id,organization_id,location_id,scope_type,scope_id,reason
) values($1,$2,$3,$4,$5,$6)`,
[id,input.organizationId,input.locationId,input.scopeType,input.scopeId ?? null,input.reason],
);
return { id, coalesced: false };
});
}
async runBatch(limit = 10) {
assertCostingSafeInteger(limit, 'costing.batch_input_invalid', 1, 100);
const jobs = await this.database.query<any>(
`with claim as (
select id
from cost_recalculation_jobs
where status='pending'
order by created_at
limit $1
for update skip locked
)
update cost_recalculation_jobs j
set status='running',started_at=now(),attempts=j.attempts+1
from claim
where j.id=claim.id
returning j.*`,
[limit],
);
let processed = 0;
for (const job of jobs) {
try {
await this.processJob(job);
await this.database.query(
`update cost_recalculation_jobs
set status='completed',completed_at=now()
where id=$1`,
[job.id],
);
processed++;
} catch (error) {
await this.database.query(
`update cost_recalculation_jobs
set status='failed',completed_at=now(),error=$2
where id=$1`,
[job.id,String(error)],
);
await this.markScopeStale(job);
}
}
return { processed };
}
private async processJob(job: any) {
const cfg = await this.config.get({
organizationId: job.organization_id,
locationId: job.location_id,
});
if (job.scope_type === 'stock_item') {
await this.recalculateStockGraph(job.organization_id,job.location_id,job.scope_id,job.reason,cfg);
return;
}
if (job.scope_type === 'ingredient') {
await this.recalculateIngredientGraph(job.organization_id,job.location_id,job.scope_id,cfg);
return;
}
if (job.scope_type === 'recipe') {
await this.recalculateRecipeGraph(job.organization_id,job.location_id,job.scope_id,cfg);
return;
}
if (job.scope_type === 'product') {
await this.margins.recalculate({
organizationId: job.organization_id,
locationId: job.location_id,
productId: job.scope_id,
costMethod: cfg.method,
currencyCode: cfg.currency_code,
});
return;
}
const stockItems = await this.database.query<{ id: string }>(
`select id from stock_items
where organization_id=$1 and is_active=true`,
[job.organization_id],
);
for (const stock of stockItems) {
await this.recalculateStockGraph(job.organization_id,job.location_id,stock.id,job.reason,cfg);
}
}
private async recalculateStockGraph(
organizationId: string,
locationId: string,
stockItemId: string,
reason: string,
cfg: any,
) {
await this.stockCosts.recalculate({organizationId,locationId,stockItemId,reason});
const ingredients = await this.database.query<{ ingredient_id: string }>(
`select ingredient_id
from stock_items
where id=$1 and organization_id=$2 and ingredient_id is not null`,
[stockItemId,organizationId],
);
for (const i of ingredients) {
await this.recalculateIngredientGraph(organizationId,locationId,i.ingredient_id,cfg);
}
}
private async recalculateIngredientGraph(
organizationId: string,
locationId: string,
ingredientId: string,
cfg: any,
) {
await this.recipeCosts.recalculateIngredient({
organizationId,locationId,ingredientId,
costMethod: cfg.method,currencyCode: cfg.currency_code,
});
const recipes = await this.database.query<{ recipe_id: string }>(
`select distinct rv.recipe_id
from recipe_components rc
join recipe_versions rv on rv.id=rc.recipe_version_id
join recipes r on r.id=rv.recipe_id and r.organization_id=$2
where rc.ingredient_id=$1`,
[ingredientId,organizationId],
);
for (const recipe of recipes) {
await this.recalculateRecipeGraph(organizationId,locationId,recipe.recipe_id,cfg);
}
}
private async recalculateRecipeGraph(
organizationId: string,
locationId: string,
recipeId: string,
cfg: any,
) {
const versions = await this.database.query<{ id: string }>(
`select id
from recipe_versions rv
join recipes r on r.id=rv.recipe_id and r.organization_id=$2
where rv.recipe_id=$1 and rv.status='published'
order by version_number desc limit 1`,
[recipeId,organizationId],
);
if (!versions[0]) return;
await this.recipeCosts.recalculateRecipe({
organizationId,locationId,recipeId,recipeVersionId:versions[0].id,
costMethod:cfg.method,currencyCode:cfg.currency_code,
includeWasteFactor: cfg.include_waste_factor,
});
const modifiers = await this.database.query<{ id: string }>(
`select id from modifiers where recipe_id=$1`,
[recipeId],
);
for (const modifier of modifiers) {
await this.modifierCosts.recalculate({
organizationId,
locationId,
modifierId: modifier.id,
currencyCode: cfg.currency_code,
});
}
const products = await this.database.query<{ id: string }>(
`select id from products
where organization_id=$1 and recipe_id=$2`,
[organizationId,recipeId],
);
for (const product of products) {
await this.margins.recalculate({
organizationId,locationId,productId:product.id,
costMethod:cfg.method,currencyCode:cfg.currency_code,
});
}
}
private async markScopeStale(job: any) {
const tables:Record<string,[string,string]> = {
stock_item:['stock_item_cost_projections','stock_item_id'],
ingredient:['ingredient_cost_projections','ingredient_id'],
recipe:['recipe_cost_projections','recipe_id'],
product:['product_margin_projections','product_id'],
};
const table=tables[String(job.scope_type)];
if (!table) return;
await this.database.query(
`update ${table[0]}
set quality_state='stale',stale_at=now()
where organization_id=$1 and location_id=$2 and ${table[1]}=$3`,
[job.organization_id,job.location_id,job.scope_id],
);
}
}