apps/backend/src/modules/menu/workers/allergen-projection.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { ProductAllergenProjectionService } from '../application/product-allergen-projection.service';
@Injectable()
export class AllergenProjectionWorker {
constructor(
private readonly database: DatabaseService,
private readonly projection: ProductAllergenProjectionService,
) {}
async enqueueProductsForRecipe(organizationId: string, recipeId: string): Promise<void> {
await this.database.query(
`insert into menu_projection_jobs(organization_id, job_type, entity_id)
select $1, 'product_allergens', p.id
from products p
where p.organization_id=$1 and p.recipe_id=$2
on conflict do nothing`,
[organizationId, recipeId],
);
}
async enqueueProductsForIngredient(
organizationId: string,
ingredientId: string,
): Promise<void> {
await this.database.query(
`with recursive affected_recipes(recipe_id) as (
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
where rc.ingredient_id=$2 and r.organization_id=$1
union
select parent_rv.recipe_id
from affected_recipes ar
join recipe_components parent_rc on parent_rc.subrecipe_id=ar.recipe_id
join recipe_versions parent_rv on parent_rv.id=parent_rc.recipe_version_id
)
insert into menu_projection_jobs(organization_id, job_type, entity_id)
select distinct $1, 'product_allergens', p.id
from products p
join affected_recipes ar on ar.recipe_id=p.recipe_id
where p.organization_id=$1
on conflict do nothing`,
[organizationId, ingredientId],
);
}
async runBatch(limit = 50): Promise<number> {
const jobs = await this.database.query<{ id: string; entity_id: string }>(
`update menu_projection_jobs
set status='running'
where id in (
select id
from menu_projection_jobs
where status='pending'
order by created_at
for update skip locked
limit $1
)
returning id, entity_id`,
[limit],
);
for (const job of jobs) {
try {
await this.projection.rebuildProduct(job.entity_id);
await this.database.query(
`update menu_projection_jobs
set status='completed', processed_at=now(), last_error=null
where id=$1`,
[job.id],
);
} catch (error) {
await this.database.query(
`update menu_projection_jobs
set status='failed', last_error=$2
where id=$1`,
[job.id, error instanceof Error ? error.message : 'unknown'],
);
}
}
return jobs.length;
}
}