apps/backend/src/modules/accounting/workers/accounting-transfer.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { createHash } from 'node:crypto';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { AccountingProviderRegistry } from '../providers/accounting-provider.registry';
@Injectable()
export class AccountingTransferWorker {
constructor(
private readonly database:DatabaseService,
private readonly providers:AccountingProviderRegistry,
) {}
async queueInvoice(input:{
organizationId:string;connectionId:string;invoiceId:string;
}) {
const hash=createHash('sha256').update(`${input.connectionId}:${input.invoiceId}`).digest('hex');
await this.database.query(
`insert into accounting_transfer_events(
id,organization_id,integration_connection_id,source_type,source_id,provider,
status,request_hash
)
select gen_random_uuid(),$1,$2,'invoice',$3,provider,'queued',$4
from integration_connections
where id=$2 and organization_id=$1 and category='accounting'
on conflict(integration_connection_id,source_type,source_id) do nothing`,
[input.organizationId,input.connectionId,input.invoiceId,hash],
);
return {queued:true};
}
async runBatch(limit=50) {
const events=await this.database.query<any>(
`with claim as (
select ate.id,ic.credential_reference,ic.config_json,ic.location_id
from accounting_transfer_events ate
join integration_connections ic on ic.id=ate.integration_connection_id
where ate.status in ('queued','failed')
order by ate.created_at
limit $1
for update of ate skip locked
)
update accounting_transfer_events ate
set status='processing'
from claim
where ate.id=claim.id
returning ate.*,claim.credential_reference,claim.config_json,claim.location_id`,
[limit],
);
let completed=0;
for(const e of events) {
const provider=this.providers.get(e.provider);
if(e.source_type!=='invoice' || !provider.transferInvoice) {
await this.database.query(
`update accounting_transfer_events
set status='failed',last_error='provider_transfer_unsupported'
where id=$1 and organization_id=$2
and (integration_connection_id=$3 or (integration_connection_id is null and $3::uuid is null))`,
[e.id,e.organization_id,e.integration_connection_id],
);
continue;
}
const invoice=await this.database.query<any>(
`select i.*,c.name as customer_name,c.email as customer_email,c.address_json,
c.vat_id,c.buyer_reference as customer_buyer_reference
from invoices i join invoice_customers c on c.id=i.invoice_customer_id
where i.id=$1 and i.organization_id=$2 and i.status='finalized'`,
[e.source_id,e.organization_id],
);
if(!invoice[0]) {
await this.database.query(
`update accounting_transfer_events
set status='failed',last_error='invoice_not_finalized'
where id=$1 and organization_id=$2
and (integration_connection_id=$3 or (integration_connection_id is null and $3::uuid is null))`,
[e.id,e.organization_id,e.integration_connection_id],
);
continue;
}
try {
const result=await provider.transferInvoice({
connectionId:e.integration_connection_id,
organizationId:e.organization_id,
locationId:e.location_id,
credentialReference:e.credential_reference,
config:e.config_json??{},
},{
idempotencyKey:`accounting:${e.id}`,
invoice:invoice[0],
});
await this.database.query(
`update accounting_transfer_events
set status=$1,external_id=$2,response_safe_json=$3::jsonb,
completed_at=case when $1='completed' then now() else null end
where id=$4 and organization_id=$5
and (integration_connection_id=$6 or (integration_connection_id is null and $6::uuid is null))`,
[
result.status==='completed'?'completed':result.status==='uncertain'?'uncertain':'processing',
result.externalId,JSON.stringify({providerStatus:result.status}),e.id,e.organization_id,e.integration_connection_id,
],
);
if(result.status==='completed') completed++;
} catch(error:any) {
await this.database.query(
`update accounting_transfer_events
set status='failed',last_error=$1
where id=$2 and organization_id=$3
and (integration_connection_id=$4 or (integration_connection_id is null and $4::uuid is null))`,
[String(error?.message??error).slice(0,1000),e.id,e.organization_id,e.integration_connection_id],
);
}
}
return {processed:events.length,completed};
}
}