import { Provide, Inject } from '@midwayjs/core';
|
import { InjectEntityModel } from '@midwayjs/typeorm';
|
import { Repository, In, LessThan } from 'typeorm';
|
import { ReportJoblogEntity } from '../entity/joblog';
|
import { TaskServiceXinYueGps } from './taskservicexinyuegps';
|
|
/**
|
* 芯跃三轮车定位实时同步执行器
|
* 负责 joblog 状态机与同步周期节流，具体拉数由 TaskServiceXinYueGps 完成
|
*/
|
@Provide()
|
export class ExecutorXinYueVehicleGps {
|
@InjectEntityModel(ReportJoblogEntity)
|
reportJoblogEntity: Repository<ReportJoblogEntity>;
|
|
@Inject()
|
taskServiceXinYueGps: TaskServiceXinYueGps;
|
|
async execute() {
|
const pendingJobs = await this.reportJoblogEntity.find({
|
where: {
|
status: In(['INIT', 'FAILED']),
|
jobName: 'XINYUE_VEHICLE_GPS',
|
},
|
order: {
|
dataDate: 'ASC',
|
},
|
});
|
|
for (const job of pendingJobs) {
|
const intervalSeconds = await this.taskServiceXinYueGps.getSyncIntervalSeconds();
|
if (this.shouldSkipByInterval(job, intervalSeconds)) {
|
continue;
|
}
|
|
const lockResult = await this.reportJoblogEntity.update(
|
{ id: job.id, status: job.status },
|
{ status: 'RUNNING', startTime: new Date() }
|
);
|
|
if (lockResult.affected === 0) {
|
continue;
|
}
|
|
try {
|
const result = await this.taskServiceXinYueGps.syncRealtimeData();
|
await this.reportJoblogEntity.update(job.id, {
|
status: 'SUCCESS',
|
recordsProcessed: result.recordsProcessed,
|
endTime: new Date(),
|
errorMessage: result.errorMessage || '',
|
});
|
} catch (error) {
|
await this.reportJoblogEntity.update(job.id, {
|
status: 'FAILED',
|
endTime: new Date(),
|
errorMessage: this.stringifyError(error),
|
failedCount: () => 'failedCount + 1',
|
});
|
}
|
}
|
}
|
|
/**
|
* 将卡住的 RUNNING 任务洗回 INIT，避免进程中断后永远不跑
|
*/
|
async recoverStuckJobs() {
|
const expireAt = new Date(Date.now() - 10 * 60 * 1000);
|
await this.reportJoblogEntity.update(
|
{
|
jobName: 'XINYUE_VEHICLE_GPS',
|
status: 'RUNNING',
|
startTime: LessThan(expireAt),
|
},
|
{ status: 'INIT' }
|
);
|
}
|
|
/**
|
* 距上次结束未到同步周期则跳过，避免后台 cron 比参数更密时刷接口
|
*/
|
private shouldSkipByInterval(job: ReportJoblogEntity, intervalSeconds: number): boolean {
|
if (!intervalSeconds || intervalSeconds <= 0) {
|
return false;
|
}
|
if (!job.endTime) {
|
return false;
|
}
|
const elapsed = Date.now() - new Date(job.endTime).getTime();
|
return elapsed < intervalSeconds * 1000;
|
}
|
|
private stringifyError(error: any): string {
|
if (error instanceof Error) {
|
return error.stack ?? error.message;
|
}
|
if (typeof error === 'string') {
|
return error;
|
}
|
try {
|
return JSON.stringify(error);
|
} catch {
|
return String(error);
|
}
|
}
|
}
|
