apps/backend/src/modules/notifications/workers/notification-delivery.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { DevEmailProvider } from '../providers/dev-email.provider';
import { NotificationPolicyService } from '../application/notification-policy.service';
@Injectable()
export class NotificationDeliveryWorker {
constructor(
private readonly database:DatabaseService,
private readonly email:DevEmailProvider,
private readonly policy:NotificationPolicyService,
) {}
async runBatch(limit=50) {
const rows=await this.database.query<any>(
`with claim as (
select nd.id,n.user_id,n.title_key,n.body_key,n.params_json,n.severity,
ua.email,np.quiet_hours_policy
from notification_deliveries nd
join notifications n on n.id=nd.notification_id
join user_accounts ua on ua.id=n.user_id
left join notification_preferences np
on np.organization_id=n.organization_id
and np.user_id=n.user_id
and np.notification_type=n.notification_type
where nd.status in ('queued','failed')
and coalesce(nd.next_attempt_at,now())<=now()
and nd.attempt_count<10
order by nd.created_at
limit $1
for update of nd skip locked
)
update notification_deliveries nd
set status='sending',attempt_count=nd.attempt_count+1,last_attempt_at=now()
from claim
where nd.id=claim.id
returning nd.*,claim.user_id,claim.title_key,claim.body_key,claim.params_json,claim.severity,claim.email,claim.quiet_hours_policy`,
[limit],
);
let sent=0;
for(const row of rows) {
if(!['urgent','critical'].includes(row.severity) &&
this.policy.isQuietHours(row.quiet_hours_policy)) {
await this.database.query(
`update notification_deliveries
set status='queued',next_attempt_at=now()+interval '30 minutes'
where id=$1 and status='sending'`,
[row.id],
);
continue;
}
try {
if(row.channel==='in_app') {
await this.database.query(
`update notification_deliveries set status='sent' where id=$1 and status='sending'`,
[row.id],
);
sent++;
continue;
}
if(row.channel==='email') {
if(!row.email) throw new Error('notifications.email_missing');
const result=await this.email.sendTransactional({
to:row.email,
subject:row.title_key,
text:row.body_key,
});
await this.database.query(
`update notification_deliveries
set status='sent',provider='dev-email',provider_message_id=$1
where id=$2 and status='sending'`,
[result.providerMessageId??null,row.id],
);
sent++;
continue;
}
throw new Error('notifications.push_provider_not_configured');
} catch(error:any) {
const attempts=Number(row.attempt_count);
const dead=attempts>=10;
await this.database.query(
`update notification_deliveries
set status=$1,last_error=$2,
next_attempt_at=case when $1='failed'
then now()+make_interval(secs=>least(3600,30*power(2,$3)::int))
else null end
where id=$4 and status='sending'`,
[dead?'dead_letter':'failed',String(error?.message??error),attempts,row.id],
);
}
}
return {sent};
}
}