apps/backend/src/infrastructure/outbox/reliable-outbox.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { randomUUID } from 'node:crypto';
import { DatabaseService } from '../database/database.service';
@Injectable()
export class ReliableOutboxWorker {
constructor(private readonly database:DatabaseService) {}
async claim(limit=100) {
const claimToken=randomUUID();
return this.database.transaction(async(tx)=>{
const rows=await tx.query<any>(
`with claim as (
select id
from outbox_events
where processed_at is null and dead_lettered_at is null
and coalesce(next_attempt_at,available_at)<=now()
and (claimed_at is null or claimed_at < now()-interval '15 minutes')
order by occurred_at
limit $2
for update skip locked
)
update outbox_events o
set claimed_at=now(),claim_token=$1
from claim
where o.id=claim.id
returning o.*`,
[claimToken,limit],
);
return rows;
},{lockTimeoutMs:2000,statementTimeoutMs:5000});
}
async succeeded(id:string,claimToken?:string) {
await this.database.query(
`update outbox_events
set processed_at=now(),last_error=null,claimed_at=null,claim_token=null
where id=$1 and processed_at is null
and ($2::uuid is null or claim_token=$2)`,
[id,claimToken??null],
);
}
async failed(id:string,error:unknown,maxAttempts=10,claimToken?:string) {
const message=String((error as any)?.message??error).slice(0,1000);
const rows=await this.database.query<any>(
`update outbox_events
set attempt_count=attempt_count+1,last_error=$1,
next_attempt_at=now()+make_interval(secs=>least(3600,30*power(2,attempt_count)::int)),
dead_lettered_at=case when attempt_count+1>=$2 then now() else dead_lettered_at end,
claimed_at=null,claim_token=null
where id=$3 and processed_at is null and dead_lettered_at is null
and ($4::uuid is null or claim_token=$4)
returning attempt_count,dead_lettered_at`,
[message,maxAttempts,id,claimToken??null],
);
return rows[0]??null;
}
async stats() {
const r=await this.database.query<any>(
`select
count(*) filter(where processed_at is null and dead_lettered_at is null)::int as pending,
count(*) filter(where dead_lettered_at is not null)::int as dead_letter,
max(attempt_count)::int as max_attempts
from outbox_events`,
);
return r[0]??{pending:0,dead_letter:0,max_attempts:0};
}
}