56 lines
2.7 KiB
JavaScript
56 lines
2.7 KiB
JavaScript
'use strict';
|
|
const fs = require('node:fs'), path = require('node:path'), crypto = require('node:crypto');
|
|
// Lazy, owner-bound, single-flight tasks. Only WeChat jobs persist. Guest photos
|
|
// remain in RAM for at most five minutes; no paid work starts until requested.
|
|
module.exports = function({dir, collect, load, complete, now = Date.now}) {
|
|
const jobs = new Map();
|
|
function remember(job) {
|
|
jobs.set(job.token,job);
|
|
const timer=setTimeout(()=>{jobs.delete(job.token);job.context=null;},300000);timer.unref();
|
|
}
|
|
function save(job) {
|
|
if (!job.persist) return;
|
|
fs.mkdirSync(dir, {recursive:true, mode:0o700});
|
|
const file=path.join(dir,job.token+'.json'),tmp=file+'.tmp';
|
|
fs.writeFileSync(tmp,JSON.stringify({token:job.token,owner:job.owner,record:job.record,persist:true,status:job.status,views:job.views}),{mode:0o600});
|
|
fs.renameSync(tmp,file);
|
|
}
|
|
function register(owner,record,context) {
|
|
const job={token:crypto.randomBytes(24).toString('hex'),owner:owner.id,record,persist:owner.kind==='wechat',status:'pending',context};
|
|
save(job);remember(job);
|
|
return {token:job.token,status:'pending'};
|
|
}
|
|
function unavailable(code) {return {version:'independent-side-wrinkles-v1',left:{status:'unavailable',error_code:code},right:{status:'unavailable',error_code:code}};}
|
|
async function run(owner,token) {
|
|
if (!/^[a-f0-9]{48}$/.test(token||'')) return null;
|
|
let job=jobs.get(token);
|
|
if (!job) {
|
|
try {job=JSON.parse(fs.readFileSync(path.join(dir,token+'.json'),'utf8'));} catch (_) {return null;}
|
|
if (job.owner!==owner.id || owner.kind!=='wechat') return null;
|
|
remember(job);
|
|
}
|
|
if (job.owner!==owner.id) return null;
|
|
if (job.promise) return job.promise;
|
|
if (job.status==='complete') return job.views;
|
|
// A persisted running marker means a process restart interrupted the task.
|
|
// Never silently repeat potentially charged requests after a restart.
|
|
const interrupted=job.status==='running';
|
|
job.status='running';save(job);
|
|
job.promise=(async()=>{
|
|
const started=now();let views;
|
|
try {
|
|
const context=job.context || (job.persist && load(job.owner,job.record));
|
|
views=interrupted?unavailable('WRINKLE_JOB_INTERRUPTED'):context?await collect(context):unavailable('WRINKLE_JOB_EXPIRED');
|
|
} catch (_) {views=unavailable('WRINKLE_JOB_FAILED');}
|
|
job.context=null;job.views=views;job.status='complete';
|
|
// Save the completed cache BEFORE patching snapshots; duplicates never
|
|
// re-run paid work, even if a snapshot or admin update fails.
|
|
save(job);
|
|
if (job.persist) {try {complete(job.owner,job.record,views,now()-started);} catch (_) {}}
|
|
return views;
|
|
})();
|
|
return job.promise;
|
|
}
|
|
return {register,run};
|
|
};
|