551 lines
10 KiB
JavaScript
551 lines
10 KiB
JavaScript
const db = require('../../db/knex');
|
|
|
|
const loc = require('../geo');
|
|
const getElevation = require('./elevation');
|
|
|
|
function serializeError(error) {
|
|
const value =
|
|
error?.stack
|
|
|| error?.message
|
|
|| String(error);
|
|
|
|
return String(value).slice(0, 16000);
|
|
}
|
|
|
|
function buildCacheKey(lat, lon) {
|
|
return (
|
|
`${Number(lat).toFixed(4)}:`
|
|
+ `${Number(lon).toFixed(4)}`
|
|
);
|
|
}
|
|
|
|
function parseLocation(value) {
|
|
if (!value) {
|
|
return null;
|
|
}
|
|
|
|
if (typeof value === 'object') {
|
|
return value;
|
|
}
|
|
|
|
try {
|
|
return JSON.parse(value);
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function serializeLocation(value) {
|
|
if (!value) {
|
|
return null;
|
|
}
|
|
|
|
if (typeof value === 'string') {
|
|
return value;
|
|
}
|
|
|
|
return JSON.stringify(value);
|
|
}
|
|
|
|
async function claimNextPendingJob() {
|
|
return db.transaction(async (trx) => {
|
|
const job = await trx('geocode_jobs')
|
|
.where({ status: 'pending' })
|
|
.orderBy('created_at', 'asc')
|
|
.orderBy('id', 'asc')
|
|
.first();
|
|
|
|
if (!job) {
|
|
return null;
|
|
}
|
|
|
|
const updated = await trx('geocode_jobs')
|
|
.where({
|
|
id: job.id,
|
|
status: 'pending',
|
|
})
|
|
.update({
|
|
status: 'processing',
|
|
attempts: trx.raw('attempts + 1'),
|
|
last_error: null,
|
|
started_at: trx.fn.now(),
|
|
completed_at: null,
|
|
updated_at: trx.fn.now(),
|
|
});
|
|
|
|
if (updated !== 1) {
|
|
return null;
|
|
}
|
|
|
|
return trx('geocode_jobs')
|
|
.where({ id: job.id })
|
|
.first();
|
|
});
|
|
}
|
|
|
|
async function completeJob(jobId) {
|
|
const updated = await db('geocode_jobs')
|
|
.where({
|
|
id: jobId,
|
|
status: 'processing',
|
|
})
|
|
.update({
|
|
status: 'done',
|
|
last_error: null,
|
|
completed_at: db.fn.now(),
|
|
updated_at: db.fn.now(),
|
|
});
|
|
|
|
if (updated !== 1) {
|
|
throw new Error(
|
|
`Impossibile completare il job id=${jobId}`,
|
|
);
|
|
}
|
|
|
|
return 'done';
|
|
}
|
|
|
|
async function failJob(jobId, error) {
|
|
const updated = await db('geocode_jobs')
|
|
.where({
|
|
id: jobId,
|
|
status: 'processing',
|
|
})
|
|
.update({
|
|
status: 'error',
|
|
last_error: serializeError(error),
|
|
completed_at: db.fn.now(),
|
|
updated_at: db.fn.now(),
|
|
});
|
|
|
|
if (updated !== 1) {
|
|
console.error(
|
|
'[geocodeWorker] impossibile registrare errore:',
|
|
jobId,
|
|
);
|
|
|
|
return 'missing';
|
|
}
|
|
|
|
return 'error';
|
|
}
|
|
|
|
async function loadEntity(job) {
|
|
if (job.entity_type === 'photo') {
|
|
return db('photos')
|
|
.select([
|
|
'id',
|
|
'lat',
|
|
'lon',
|
|
'alt',
|
|
'location',
|
|
])
|
|
.where({ id: job.entity_id })
|
|
.whereNull('deleted_at')
|
|
.first();
|
|
}
|
|
|
|
if (job.entity_type === 'activity') {
|
|
return db('activities')
|
|
.select([
|
|
'id',
|
|
'center_latitude',
|
|
'center_longitude',
|
|
'location',
|
|
])
|
|
.where({ id: job.entity_id })
|
|
.whereNull('deleted_at')
|
|
.first();
|
|
}
|
|
|
|
throw new Error(
|
|
`Tipo entità non valido: ${job.entity_type}`,
|
|
);
|
|
}
|
|
|
|
async function loadCache(cacheKey) {
|
|
return db('geocode_cache')
|
|
.where({ cache_key: cacheKey })
|
|
.first();
|
|
}
|
|
|
|
async function saveCache({
|
|
cacheKey,
|
|
lat,
|
|
lon,
|
|
alt,
|
|
location,
|
|
}) {
|
|
const existing = await loadCache(cacheKey);
|
|
|
|
const finalAlt =
|
|
alt != null
|
|
? Number(alt)
|
|
: existing?.alt ?? null;
|
|
|
|
const finalLocation =
|
|
serializeLocation(location)
|
|
|| existing?.location
|
|
|| null;
|
|
|
|
const now = db.fn.now();
|
|
|
|
await db('geocode_cache')
|
|
.insert({
|
|
cache_key: cacheKey,
|
|
lat,
|
|
lon,
|
|
alt: finalAlt,
|
|
location: finalLocation,
|
|
hits: existing?.hits ?? 0,
|
|
created_at:
|
|
existing?.created_at || now,
|
|
updated_at: now,
|
|
})
|
|
.onConflict('cache_key')
|
|
.merge({
|
|
lat,
|
|
lon,
|
|
alt: finalAlt,
|
|
location: finalLocation,
|
|
updated_at: now,
|
|
});
|
|
}
|
|
|
|
async function recordCacheHit(cacheKey) {
|
|
await db('geocode_cache')
|
|
.where({ cache_key: cacheKey })
|
|
.update({
|
|
hits: db.raw('hits + 1'),
|
|
updated_at: db.fn.now(),
|
|
});
|
|
}
|
|
|
|
async function updateEntity({
|
|
job,
|
|
alt,
|
|
location,
|
|
}) {
|
|
const locationJson =
|
|
serializeLocation(location);
|
|
|
|
if (job.entity_type === 'photo') {
|
|
const updated = await db('photos')
|
|
.where({ id: job.entity_id })
|
|
.whereNull('deleted_at')
|
|
.update({
|
|
alt: alt ?? null,
|
|
location: locationJson,
|
|
updated_at: db.fn.now(),
|
|
});
|
|
|
|
if (updated !== 1) {
|
|
throw new Error(
|
|
`Foto non trovata: ${job.entity_id}`,
|
|
);
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
if (job.entity_type === 'activity') {
|
|
const updated = await db('activities')
|
|
.where({ id: job.entity_id })
|
|
.whereNull('deleted_at')
|
|
.update({
|
|
location: locationJson,
|
|
updated_at: db.fn.now(),
|
|
});
|
|
|
|
if (updated !== 1) {
|
|
throw new Error(
|
|
`Attività non trovata: ${job.entity_id}`,
|
|
);
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
throw new Error(
|
|
`Tipo entità non valido: ${job.entity_type}`,
|
|
);
|
|
}
|
|
|
|
async function processClaimedJob(job) {
|
|
if (!job?.id || !job?.entity_id) {
|
|
throw new Error('Job geocode non valido');
|
|
}
|
|
|
|
const workerStartedAt = Date.now();
|
|
|
|
console.log(
|
|
'[geocodeWorker] processing',
|
|
{
|
|
job_id: job.id,
|
|
entity_type: job.entity_type,
|
|
entity_id: job.entity_id,
|
|
attempt: job.attempts,
|
|
},
|
|
);
|
|
|
|
try {
|
|
const lat = Number(job.lat);
|
|
const lon = Number(job.lon);
|
|
|
|
if (
|
|
!Number.isFinite(lat)
|
|
|| !Number.isFinite(lon)
|
|
|| lat < -90
|
|
|| lat > 90
|
|
|| lon < -180
|
|
|| lon > 180
|
|
) {
|
|
throw new Error(
|
|
`Coordinate non valide: lat=${job.lat}, lon=${job.lon}`,
|
|
);
|
|
}
|
|
|
|
const entity = await loadEntity(job);
|
|
|
|
if (!entity) {
|
|
throw new Error(
|
|
`Entità non trovata: `
|
|
+ `${job.entity_type}/${job.entity_id}`,
|
|
);
|
|
}
|
|
|
|
const cacheKey = buildCacheKey(lat, lon);
|
|
const cached = await loadCache(cacheKey);
|
|
|
|
const altitudeFromCache =
|
|
job.entity_type === 'photo'
|
|
&& entity.alt == null
|
|
&& cached?.alt != null;
|
|
|
|
const locationFromCache =
|
|
!parseLocation(entity.location)
|
|
&& Boolean(
|
|
parseLocation(cached?.location)
|
|
);
|
|
|
|
const cacheUsed =
|
|
altitudeFromCache
|
|
|| locationFromCache;
|
|
|
|
let altitude =
|
|
job.entity_type === 'photo'
|
|
? entity.alt ?? cached?.alt ?? null
|
|
: null;
|
|
|
|
let location =
|
|
parseLocation(entity.location)
|
|
|| parseLocation(cached?.location)
|
|
|| null;
|
|
|
|
if (cacheUsed) {
|
|
await recordCacheHit(cacheKey);
|
|
}
|
|
|
|
if (
|
|
job.entity_type === 'photo'
|
|
&& altitude == null
|
|
) {
|
|
altitude = await getElevation(
|
|
lat,
|
|
lon,
|
|
);
|
|
}
|
|
|
|
if (!location) {
|
|
location = await loc(
|
|
lon,
|
|
lat,
|
|
);
|
|
}
|
|
|
|
if (!location) {
|
|
throw new Error(
|
|
`Reverse geocoding senza risultato: `
|
|
+ `lat=${lat}, lon=${lon}`,
|
|
);
|
|
}
|
|
|
|
await saveCache({
|
|
cacheKey,
|
|
lat,
|
|
lon,
|
|
alt: altitude,
|
|
location,
|
|
});
|
|
|
|
await updateEntity({
|
|
job,
|
|
alt: altitude,
|
|
location,
|
|
});
|
|
|
|
const finalStatus =
|
|
await completeJob(job.id);
|
|
|
|
const summary = {
|
|
status: finalStatus,
|
|
job_id: job.id,
|
|
entity_type: job.entity_type,
|
|
entity_id: job.entity_id,
|
|
attempt: job.attempts,
|
|
cache_key: cacheKey,
|
|
cache_used: cacheUsed,
|
|
altitude_from_cache: altitudeFromCache,
|
|
location_from_cache: locationFromCache,
|
|
altitude:
|
|
job.entity_type === 'photo'
|
|
? altitude
|
|
: null,
|
|
worker_ms:
|
|
Date.now() - workerStartedAt,
|
|
};
|
|
|
|
console.log(
|
|
'[geocodeWorker] done',
|
|
summary,
|
|
);
|
|
|
|
return summary;
|
|
} catch (error) {
|
|
const finalStatus = await failJob(
|
|
job.id,
|
|
error,
|
|
);
|
|
|
|
const summary = {
|
|
status: finalStatus,
|
|
job_id: job.id,
|
|
entity_type: job.entity_type,
|
|
entity_id: job.entity_id,
|
|
attempt: job.attempts,
|
|
error: serializeError(error),
|
|
worker_ms:
|
|
Date.now() - workerStartedAt,
|
|
};
|
|
|
|
console.error(
|
|
'[geocodeWorker] error',
|
|
summary,
|
|
);
|
|
|
|
return summary;
|
|
}
|
|
}
|
|
|
|
async function processNextJob() {
|
|
const job = await claimNextPendingJob();
|
|
|
|
if (!job) {
|
|
return null;
|
|
}
|
|
|
|
return processClaimedJob(job);
|
|
}
|
|
|
|
async function runWorker(options = {}) {
|
|
const maxJobs =
|
|
options.maxJobs == null
|
|
? null
|
|
: Number(options.maxJobs);
|
|
|
|
if (
|
|
maxJobs !== null
|
|
&& (
|
|
!Number.isInteger(maxJobs)
|
|
|| maxJobs <= 0
|
|
)
|
|
) {
|
|
throw new Error(
|
|
'maxJobs deve essere un intero positivo',
|
|
);
|
|
}
|
|
|
|
const shouldStop =
|
|
typeof options.shouldStop === 'function'
|
|
? options.shouldStop
|
|
: () => false;
|
|
|
|
const results = [];
|
|
|
|
while (
|
|
!shouldStop()
|
|
&& (
|
|
maxJobs === null
|
|
|| results.length < maxJobs
|
|
)
|
|
) {
|
|
const result = await processNextJob();
|
|
|
|
if (!result) {
|
|
break;
|
|
}
|
|
|
|
results.push(result);
|
|
}
|
|
|
|
return {
|
|
jobs_processed: results.length,
|
|
jobs_succeeded: results.filter(
|
|
(result) => result.status === 'done',
|
|
).length,
|
|
jobs_failed: results.filter(
|
|
(result) => result.status === 'error',
|
|
).length,
|
|
stopped: shouldStop(),
|
|
results,
|
|
};
|
|
}
|
|
|
|
async function resetStaleProcessingJobs(
|
|
staleMinutes = 30,
|
|
) {
|
|
const minutes = Number(staleMinutes);
|
|
|
|
if (
|
|
!Number.isFinite(minutes)
|
|
|| minutes <= 0
|
|
) {
|
|
throw new Error(
|
|
'staleMinutes deve essere positivo',
|
|
);
|
|
}
|
|
|
|
const modifier =
|
|
`-${Math.floor(minutes)} minutes`;
|
|
|
|
return db('geocode_jobs')
|
|
.where({ status: 'processing' })
|
|
.where(
|
|
'started_at',
|
|
'<',
|
|
db.raw(
|
|
"datetime('now', ?)",
|
|
[modifier],
|
|
),
|
|
)
|
|
.update({
|
|
status: 'pending',
|
|
last_error:
|
|
'Job recuperato dopo arresto del worker',
|
|
started_at: null,
|
|
completed_at: null,
|
|
updated_at: db.fn.now(),
|
|
});
|
|
}
|
|
|
|
module.exports = {
|
|
serializeError,
|
|
buildCacheKey,
|
|
claimNextPendingJob,
|
|
completeJob,
|
|
failJob,
|
|
processClaimedJob,
|
|
processNextJob,
|
|
runWorker,
|
|
resetStaleProcessingJobs,
|
|
};
|