apps/backend/src/modules/notifications/application/notification.service.tsTypeScript
import { Injectable } from '@nestjs/common';
import { randomUUID } from 'node:crypto';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { DomainConflictError } from '../../../common/errors/domain-conflict.error';
import { NotificationPolicyService } from './notification-policy.service';
import {
assertNotificationCreateInput,
assertNotificationPreferenceInput,
} from './notification-input-validation';
@Injectable()
export class NotificationService {
constructor(
private readonly database:DatabaseService,
private readonly policy:NotificationPolicyService,
) {}
async create(input:unknown) {
assertNotificationCreateInput(input);
return this.database.transaction(async(tx)=>{
if(input.deduplicationKey) {
await tx.query(
`select pg_advisory_xact_lock(hashtext($1))`,
[`notification-dedup:${input.organizationId}:${input.userId}:${input.deduplicationKey}`],
);
const replay=await tx.query<any>(
`select id,status
from notifications
where organization_id=$1 and user_id=$2 and deduplication_key=$3`,
[input.organizationId,input.userId,input.deduplicationKey],
);
if(replay[0]) return {...replay[0],deduplicated:true};
}
const policy=await this.policy.shouldCreate({
organizationId:input.organizationId,
userId:input.userId,
notificationType:input.notificationType,
throttleKey:input.throttleKey,
severity:input.severity??'info',
throttleMinutes:input.throttleMinutes,
},tx);
if(!policy.allowed) return {suppressed:true,reason:policy.reason};
const id=randomUUID();
try {
await tx.query(
`insert into notifications(
id,organization_id,user_id,location_id,notification_type,
title_key,body_key,params_json,severity,related_entity_type,
related_entity_id,deduplication_key
) values($1,$2,$3,$4,$5,$6,$7,$8::jsonb,$9,$10,$11,$12)`,
[
id,input.organizationId,input.userId,input.locationId??null,input.notificationType,
input.titleKey,input.bodyKey,JSON.stringify(input.params??{}),input.severity??'info',
input.relatedEntityType??null,input.relatedEntityId??null,input.deduplicationKey??null,
],
);
} catch(error:any) {
if(input.deduplicationKey&&String(error?.code)==='23505') {
const replay=await tx.query<any>(
`select id,status
from notifications
where organization_id=$1 and user_id=$2 and deduplication_key=$3`,
[input.organizationId,input.userId,input.deduplicationKey],
);
if(replay[0]) return {...replay[0],deduplicated:true};
}
throw error;
}
await this.policy.markCreated({
organizationId:input.organizationId,
userId:input.userId,
notificationType:input.notificationType,
throttleKey:input.throttleKey,
},tx);
const pref=await tx.query<any>(
`select in_app_enabled,email_enabled,push_enabled
from notification_preferences
where organization_id=$1 and user_id=$2 and notification_type=$3`,
[input.organizationId,input.userId,input.notificationType],
);
const p=pref[0]??{in_app_enabled:true,email_enabled:false,push_enabled:false};
const channels:string[]=[];
if(p.in_app_enabled) channels.push('in_app');
if(p.email_enabled) channels.push('email');
if(p.push_enabled) channels.push('push');
for(const channel of channels) {
await tx.query(
`insert into notification_deliveries(
id,notification_id,channel,status,next_attempt_at
) values($1,$2,$3,'queued',now())
on conflict(notification_id,channel) do nothing`,
[randomUUID(),id,channel],
);
}
return {id,status:'unread',deduplicated:false};
},{lockTimeoutMs:3000,statementTimeoutMs:10000});
}
async list(input:{organizationId:string;userId:string;locationId?:string}) {
if (input.locationId) {
return this.database.query<any>(
`select *
from notifications
where organization_id=$1 and user_id=$2 and status<>'archived'
and (location_id=$3 or location_id is null)
order by created_at desc limit 200`,
[input.organizationId,input.userId,input.locationId],
);
}
return this.database.query<any>(
`select *
from notifications
where organization_id=$1 and user_id=$2 and status<>'archived'
order by created_at desc limit 200`,
[input.organizationId,input.userId],
);
}
async read(input:{organizationId:string;userId:string;id:string;locationId?:string}) {
const locationClause = input.locationId
? ` and (location_id=$4 or location_id is null)`
: '';
const rows=await this.database.query<any>(
`update notifications
set status='read',read_at=coalesce(read_at,now())
where id=$1 and organization_id=$2 and user_id=$3 and status='unread'${locationClause}
returning id,status`,
input.locationId
? [input.id,input.organizationId,input.userId,input.locationId]
: [input.id,input.organizationId,input.userId],
);
if(!rows[0]) throw new DomainConflictError('notifications.read_invalid');
return rows[0];
}
async readAll(input:{organizationId:string;userId:string;locationId?:string}) {
const locationClause = input.locationId
? ` and (location_id=$3 or location_id is null)`
: '';
const result=await this.database.query<any>(
`update notifications
set status='read',read_at=coalesce(read_at,now())
where organization_id=$1 and user_id=$2 and status='unread'${locationClause}
returning id`,
input.locationId
? [input.organizationId,input.userId,input.locationId]
: [input.organizationId,input.userId],
);
return {updated:result.length};
}
async preferences(input:{organizationId:string;userId:string}) {
return this.database.query<any>(
`select notification_type,in_app_enabled,email_enabled,push_enabled,quiet_hours_policy
from notification_preferences
where organization_id=$1 and user_id=$2
order by notification_type`,
[input.organizationId,input.userId],
);
}
async putPreference(input:unknown) {
assertNotificationPreferenceInput(input);
await this.database.query(
`insert into notification_preferences(
id,organization_id,user_id,notification_type,
in_app_enabled,email_enabled,push_enabled,quiet_hours_policy
) values($1,$2,$3,$4,$5,$6,$7,$8::jsonb)
on conflict(organization_id,user_id,notification_type)
do update set
in_app_enabled=excluded.in_app_enabled,
email_enabled=excluded.email_enabled,
push_enabled=excluded.push_enabled,
quiet_hours_policy=excluded.quiet_hours_policy,
updated_at=now()`,
[
randomUUID(),input.organizationId,input.userId,input.notificationType,
input.inAppEnabled??true,input.emailEnabled??false,input.pushEnabled??false,
JSON.stringify(input.quietHoursPolicy??null),
],
);
return {updated:true};
}
}