apps/backend/src/modules/accounting/workers/accounting-export.worker.tsTypeScript
import { Injectable } from '@nestjs/common';
import { DatabaseService } from '../../../infrastructure/database/database.service';
import { AccountingExportService } from '../application/accounting-export.service';
import { AccountingArtifactStorageService } from '../application/accounting-artifact-storage.service';
import { DatevCsvService } from '../application/datev-csv.service';
import { assertAccountingScope, assertAccountingText } from '../application/accounting-input-validation';
type AccountingExportRunInput = string | {
organizationId: string;
locationId: string;
exportJobId: string;
};
@Injectable()
export class AccountingExportWorker {
constructor(
private readonly database:DatabaseService,
private readonly exports:AccountingExportService,
private readonly artifacts:AccountingArtifactStorageService,
private readonly datev:DatevCsvService,
) {}
async runOne(input: AccountingExportRunInput) {
const scoped = typeof input !== 'string';
if (scoped) {
assertAccountingScope(input, 'accounting.export_input_invalid');
assertAccountingText(input.exportJobId, 200, 'accounting.export_input_invalid');
} else {
assertAccountingText(input, 200, 'accounting.export_input_invalid');
}
const jobId = typeof input === 'string' ? input : input.exportJobId;
const scopeParams = typeof input === 'string'
? [jobId]
: [jobId, input.organizationId, input.locationId];
const scopeWhere = typeof input === 'string'
? 'id=$1 and status=\'queued\''
: 'id=$1 and organization_id=$2 and location_id=$3 and status=\'queued\'';
const jobs=await this.database.query<any>(
`select * from accounting_export_jobs where ${scopeWhere}`,
scopeParams,
);
const job=jobs[0];
if(!job) return {processed:false};
const claimed=await this.database.query<any>(
`update accounting_export_jobs
set status='processing',started_at=now()
where ${scopeWhere}
returning id`,
scopeParams,
);
if(!claimed[0]) return {processed:false};
try {
await this.exports.buildRows(jobId);
const rows=await this.database.query<any>(
`select * from accounting_export_rows
where accounting_export_job_id=$1 order by row_number`,
[jobId],
);
let content:Buffer;
let fileName:string;
let artifactType:string;
if(job.export_type==='datev_csv') {
content=this.datev.render(rows,job.options_json??{});
fileName=`datev-${job.period_start}-${job.period_end}.csv`;
artifactType='datev_csv';
} else if(job.export_type==='journal_csv') {
const cell=(v:any)=>{
let s=v==null?'':String(v);
if(/^[=+\-@]/.test(s)) s=`'${s}`;
return `"${s.replace(/"/g,'""')}"`;
};
const headers=['booking_date','debit_account','credit_account','amount_minor','currency_code','tax_key','reference','booking_text'];
const lines=[headers.map(cell).join(',')];
for(const row of rows) lines.push(headers.map(h=>cell(row[h])).join(','));
content=Buffer.from(lines.join('\r\n'),'utf8');
fileName=`journal-${job.period_start}-${job.period_end}.csv`;
artifactType='journal_csv';
} else {
await this.database.query(
`update accounting_export_jobs set status='unsupported',completed_at=now(),
last_error='This export type is handled by invoice/provider-specific workflows.'
where id=$1`,
[jobId],
);
return {processed:true,status:'unsupported'};
}
const stored=await this.artifacts.store({
organizationId:job.organization_id,
locationId:job.location_id,
sourceType:'accounting_export_job',
sourceId:job.id,
artifactType,
fileName,
mimeType:'text/csv',
content,
userId:job.requested_by_user_id,
});
await this.database.query(
`update accounting_export_jobs
set status='completed',file_asset_id=$1,completed_at=now()
where id=$2`,
[stored.fileAssetId,jobId],
);
return {processed:true,status:'completed',fileAssetId:stored.fileAssetId};
} catch(error:any) {
await this.database.query(
`update accounting_export_jobs
set status='failed',completed_at=now(),last_error=$1 where id=$2`,
[String(error?.message??error).slice(0,1000),jobId],
);
return {processed:true,status:'failed'};
}
}
}