576 lines
17 KiB
JavaScript
576 lines
17 KiB
JavaScript
'use strict';
|
|
|
|
const DEFAULT_REFRESH_MODE = 'incremental_scan';
|
|
const DEFAULT_SCAN_PAGE_SIZE = 50;
|
|
const DEFAULT_DAILY_FULL_REFRESH_HOUR = 4;
|
|
const SCHEDULER_TICK_MS = 60 * 1000;
|
|
const MAX_SCAN_PAGES = 500;
|
|
|
|
const CHANGE_EVENT_TYPES = new Set([
|
|
'created',
|
|
'updated',
|
|
'comment',
|
|
'status',
|
|
'executor_changed',
|
|
]);
|
|
|
|
const REMOVE_EVENT_TYPES = new Set([
|
|
'archived',
|
|
'out_of_scope',
|
|
'deleted',
|
|
]);
|
|
|
|
function safeIsoNow() {
|
|
return new Date().toISOString();
|
|
}
|
|
|
|
function toInt(value, fallback) {
|
|
const n = parseInt(value, 10);
|
|
return Number.isFinite(n) ? n : fallback;
|
|
}
|
|
|
|
function normalizeRefreshMode(mode) {
|
|
return mode === 'webhook_daily' ? 'webhook_daily' : DEFAULT_REFRESH_MODE;
|
|
}
|
|
|
|
function normalizePageSize(value) {
|
|
const n = toInt(value, DEFAULT_SCAN_PAGE_SIZE);
|
|
return Math.max(10, Math.min(500, n));
|
|
}
|
|
|
|
function normalizeDailyHour(value) {
|
|
const n = toInt(value, DEFAULT_DAILY_FULL_REFRESH_HOUR);
|
|
return Math.max(0, Math.min(23, n));
|
|
}
|
|
|
|
function sortByUpdatedDesc(tasks) {
|
|
return [...tasks].sort((a, b) => {
|
|
const au = a?.updatedat || '';
|
|
const bu = b?.updatedat || '';
|
|
if (bu === au) return 0;
|
|
return bu > au ? 1 : -1;
|
|
});
|
|
}
|
|
|
|
function extractTaskId(task) {
|
|
return String(task?.id ?? '');
|
|
}
|
|
|
|
function extractTaskNumber(task) {
|
|
return String(task?.tasknumber ?? task?.id ?? '');
|
|
}
|
|
|
|
function extractUpdatedAt(task) {
|
|
return String(task?.updatedat ?? '');
|
|
}
|
|
|
|
function hasChangesByIdAndUpdated(oldTasks, newTasks) {
|
|
if (oldTasks.length !== newTasks.length) return true;
|
|
const oldMap = new Map(oldTasks.map((t) => [extractTaskId(t), extractUpdatedAt(t)]));
|
|
for (const task of newTasks) {
|
|
const id = extractTaskId(task);
|
|
if (!oldMap.has(id)) return true;
|
|
if (oldMap.get(id) !== extractUpdatedAt(task)) return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
function createRefreshService({ db, fetch, getSettings, sseBroadcast, logger = console }) {
|
|
db.exec(`
|
|
CREATE TABLE IF NOT EXISTS sync_state (
|
|
id INTEGER PRIMARY KEY CHECK (id = 1),
|
|
cache_version INTEGER NOT NULL DEFAULT 0,
|
|
refresh_in_progress INTEGER NOT NULL DEFAULT 0,
|
|
last_success_at TEXT,
|
|
last_error TEXT,
|
|
last_full_reconcile_at TEXT
|
|
);
|
|
|
|
INSERT OR IGNORE INTO sync_state(id, cache_version, refresh_in_progress)
|
|
VALUES(1, 0, 0);
|
|
|
|
CREATE TABLE IF NOT EXISTS tasks_index (
|
|
task_id TEXT PRIMARY KEY,
|
|
task_number TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_tasks_index_number ON tasks_index(task_number);
|
|
`);
|
|
|
|
const stmts = {
|
|
getCacheRow: db.prepare('SELECT data, updated_at FROM tasks_cache WHERE id=1'),
|
|
upsertCache: db.prepare(`
|
|
INSERT INTO tasks_cache(id,data,updated_at) VALUES(1,?,?)
|
|
ON CONFLICT(id) DO UPDATE SET data=excluded.data,updated_at=excluded.updated_at
|
|
`),
|
|
getSyncState: db.prepare(`
|
|
SELECT cache_version, refresh_in_progress, last_success_at, last_error, last_full_reconcile_at
|
|
FROM sync_state WHERE id=1
|
|
`),
|
|
setInProgress: db.prepare('UPDATE sync_state SET refresh_in_progress=? WHERE id=1'),
|
|
markError: db.prepare(`
|
|
UPDATE sync_state
|
|
SET refresh_in_progress=0, last_error=?
|
|
WHERE id=1
|
|
`),
|
|
markSuccessNoChange: db.prepare(`
|
|
UPDATE sync_state
|
|
SET refresh_in_progress=0, last_success_at=?, last_error=?
|
|
WHERE id=1
|
|
`),
|
|
markSuccessNoChangeFull: db.prepare(`
|
|
UPDATE sync_state
|
|
SET refresh_in_progress=0, last_success_at=?, last_error=?, last_full_reconcile_at=?
|
|
WHERE id=1
|
|
`),
|
|
writeStateAfterCache: db.prepare(`
|
|
UPDATE sync_state
|
|
SET cache_version=?, refresh_in_progress=0, last_success_at=?, last_error=?
|
|
WHERE id=1
|
|
`),
|
|
writeStateAfterFull: db.prepare(`
|
|
UPDATE sync_state
|
|
SET cache_version=?, refresh_in_progress=0, last_success_at=?, last_error=?, last_full_reconcile_at=?
|
|
WHERE id=1
|
|
`),
|
|
clearIndex: db.prepare('DELETE FROM tasks_index'),
|
|
insertIndex: db.prepare('INSERT INTO tasks_index(task_id, task_number, updated_at) VALUES(?,?,?)'),
|
|
};
|
|
|
|
function getSyncState() {
|
|
const row = stmts.getSyncState.get();
|
|
if (row) return row;
|
|
return {
|
|
cache_version: 0,
|
|
refresh_in_progress: 0,
|
|
last_success_at: null,
|
|
last_error: null,
|
|
last_full_reconcile_at: null,
|
|
};
|
|
}
|
|
|
|
function readCachedTasks() {
|
|
try {
|
|
const row = stmts.getCacheRow.get();
|
|
if (!row) return { tasks: [], updated_at: null };
|
|
return {
|
|
tasks: JSON.parse(row.data || '[]'),
|
|
updated_at: row.updated_at || null,
|
|
};
|
|
} catch {
|
|
return { tasks: [], updated_at: null };
|
|
}
|
|
}
|
|
|
|
function normalizeSettings() {
|
|
const s = getSettings();
|
|
return {
|
|
...s,
|
|
refresh_mode: normalizeRefreshMode(s.refresh_mode),
|
|
scan_page_size: normalizePageSize(s.scan_page_size),
|
|
daily_full_refresh_hour: normalizeDailyHour(s.daily_full_refresh_hour),
|
|
};
|
|
}
|
|
|
|
function writeCacheAndIndexes(tasks, { fullReconcile = false } = {}) {
|
|
const now = safeIsoNow();
|
|
const state = getSyncState();
|
|
const nextVersion = Number(state.cache_version || 0) + 1;
|
|
const sorted = sortByUpdatedDesc(tasks);
|
|
|
|
const tx = db.transaction(() => {
|
|
stmts.upsertCache.run(JSON.stringify(sorted), now);
|
|
stmts.clearIndex.run();
|
|
for (const task of sorted) {
|
|
const taskId = extractTaskId(task);
|
|
if (!taskId) continue;
|
|
stmts.insertIndex.run(taskId, extractTaskNumber(task), extractUpdatedAt(task) || now);
|
|
}
|
|
if (fullReconcile) {
|
|
stmts.writeStateAfterFull.run(nextVersion, now, null, now);
|
|
} else {
|
|
stmts.writeStateAfterCache.run(nextVersion, now, null);
|
|
}
|
|
});
|
|
tx();
|
|
|
|
sseBroadcast('tasks_updated', {
|
|
updated_at: now,
|
|
count: sorted.length,
|
|
cache_version: nextVersion,
|
|
});
|
|
|
|
return {
|
|
changed: true,
|
|
count: sorted.length,
|
|
updated_at: now,
|
|
cache_version: nextVersion,
|
|
};
|
|
}
|
|
|
|
function markSuccessNoChange({ fullReconcile = false } = {}) {
|
|
const now = safeIsoNow();
|
|
if (fullReconcile) {
|
|
stmts.markSuccessNoChangeFull.run(now, null, now);
|
|
} else {
|
|
stmts.markSuccessNoChange.run(now, null);
|
|
}
|
|
return {
|
|
changed: false,
|
|
count: readCachedTasks().tasks.length,
|
|
updated_at: readCachedTasks().updated_at,
|
|
cache_version: getSyncState().cache_version || 0,
|
|
};
|
|
}
|
|
|
|
function markError(error) {
|
|
const message = String(error?.message || error || 'Unknown refresh error').slice(0, 1000);
|
|
stmts.markError.run(message);
|
|
}
|
|
|
|
async function buildTasksPageFromApi(top, skip = 0) {
|
|
const s = getSettings();
|
|
if (!s.api_key) throw new Error('API ключ не задан.');
|
|
|
|
const executorFilter = [
|
|
'executorgroup eq 243025',
|
|
'executor eq 2227907',
|
|
'executor eq 525111',
|
|
'executor eq 631481',
|
|
'executor eq 2222953',
|
|
].join(' or ');
|
|
|
|
const topParam = Math.max(1, top);
|
|
const skipParam = Math.max(0, skip);
|
|
const fullFilter = `(${executorFilter})`;
|
|
const url = `https://apigw.intradesk.ru/tasklist/odata/v3/tasks?ApiKey=${s.api_key}`
|
|
+ `&$filter=${encodeURIComponent(`(${fullFilter})`)}`
|
|
+ `&$orderby=updatedat%20desc&$top=${topParam}&$skip=${skipParam}`;
|
|
|
|
logger.log(`[IntraDesk] GET top=${topParam} skip=${skipParam}`, url.replace(s.api_key, '***'));
|
|
const response = await fetch(url, { timeout: 30000 });
|
|
logger.log('[IntraDesk] Status:', response.status, response.statusText);
|
|
if (!response.ok) throw new Error(`IntraDesk вернул ${response.status}: ${response.statusText}`);
|
|
|
|
const json = await response.json();
|
|
const tasks = json.value || [];
|
|
const dictMap = {};
|
|
(json.dictionaries || []).forEach((d) => { dictMap[d.id] = d.name; });
|
|
|
|
tasks.forEach((task) => {
|
|
if (task.clientid && dictMap[task.clientid]) task._clientname = dictMap[task.clientid];
|
|
if (task.executor && dictMap[task.executor]) task._executorname = dictMap[task.executor];
|
|
if (task.executorgroup && dictMap[task.executorgroup]) task._executorgroupname = dictMap[task.executorgroup];
|
|
const statusKey = task.statusid || task.status;
|
|
if (statusKey && dictMap[statusKey]) task._statusname = dictMap[statusKey];
|
|
});
|
|
|
|
return tasks;
|
|
}
|
|
|
|
async function fetchAllTasksPaged(pageSize) {
|
|
const byId = new Map();
|
|
let skip = 0;
|
|
let pages = 0;
|
|
while (pages < MAX_SCAN_PAGES) {
|
|
const page = await buildTasksPageFromApi(pageSize, skip);
|
|
pages += 1;
|
|
if (!page.length) break;
|
|
for (const task of page) {
|
|
byId.set(extractTaskId(task), task);
|
|
}
|
|
if (page.length < pageSize) break;
|
|
skip += pageSize;
|
|
}
|
|
return sortByUpdatedDesc([...byId.values()]);
|
|
}
|
|
|
|
async function runIncrementalScan(reason = 'manual_incremental') {
|
|
const settings = normalizeSettings();
|
|
const pageSize = settings.scan_page_size;
|
|
const { tasks: existing } = readCachedTasks();
|
|
const byId = new Map(existing.map((t) => [extractTaskId(t), t]));
|
|
|
|
let changed = false;
|
|
let skip = 0;
|
|
let pages = 0;
|
|
|
|
while (pages < MAX_SCAN_PAGES) {
|
|
const page = await buildTasksPageFromApi(pageSize, skip);
|
|
pages += 1;
|
|
|
|
if (!page.length) break;
|
|
|
|
let pageChanged = false;
|
|
for (const task of page) {
|
|
const taskId = extractTaskId(task);
|
|
const previous = byId.get(taskId);
|
|
if (!previous || extractUpdatedAt(previous) !== extractUpdatedAt(task)) {
|
|
byId.set(taskId, task);
|
|
changed = true;
|
|
pageChanged = true;
|
|
}
|
|
}
|
|
|
|
if (!pageChanged) break;
|
|
if (page.length < pageSize) break;
|
|
skip += pageSize;
|
|
}
|
|
|
|
if (!changed) {
|
|
logger.log(`[Refresh] Incremental scan: no changes (${reason})`);
|
|
return markSuccessNoChange({ fullReconcile: false });
|
|
}
|
|
|
|
logger.log(`[Refresh] Incremental scan: changed (${reason})`);
|
|
return writeCacheAndIndexes([...byId.values()], { fullReconcile: false });
|
|
}
|
|
|
|
async function runFullReconcile(reason = 'manual_full_reconcile') {
|
|
const settings = normalizeSettings();
|
|
const pageSize = settings.scan_page_size;
|
|
const nextTasks = await fetchAllTasksPaged(pageSize);
|
|
const currentTasks = readCachedTasks().tasks;
|
|
|
|
if (!hasChangesByIdAndUpdated(currentTasks, nextTasks)) {
|
|
logger.log(`[Refresh] Full reconcile: no changes (${reason})`);
|
|
return markSuccessNoChange({ fullReconcile: true });
|
|
}
|
|
|
|
logger.log(`[Refresh] Full reconcile: changed (${reason})`);
|
|
return writeCacheAndIndexes(nextTasks, { fullReconcile: true });
|
|
}
|
|
|
|
function removeTaskFromCache(taskId, reason = 'webhook_remove') {
|
|
const id = String(taskId || '').trim();
|
|
if (!id) return markSuccessNoChange({ fullReconcile: false });
|
|
|
|
const { tasks: existing } = readCachedTasks();
|
|
const filtered = existing.filter((t) => extractTaskId(t) !== id);
|
|
if (filtered.length === existing.length) {
|
|
logger.log(`[Refresh] Remove task: not found ${id} (${reason})`);
|
|
return markSuccessNoChange({ fullReconcile: false });
|
|
}
|
|
|
|
logger.log(`[Refresh] Remove task: removed ${id} (${reason})`);
|
|
return writeCacheAndIndexes(filtered, { fullReconcile: false });
|
|
}
|
|
|
|
function shouldRunDailyFull(settings, syncState, now = new Date()) {
|
|
const hour = settings.daily_full_refresh_hour;
|
|
if (now.getHours() < hour) return false;
|
|
|
|
const last = syncState.last_full_reconcile_at ? new Date(syncState.last_full_reconcile_at) : null;
|
|
if (!last || Number.isNaN(last.getTime())) return true;
|
|
|
|
const dayKeyNow = `${now.getFullYear()}-${now.getMonth()}-${now.getDate()}`;
|
|
const dayKeyLast = `${last.getFullYear()}-${last.getMonth()}-${last.getDate()}`;
|
|
return dayKeyNow !== dayKeyLast;
|
|
}
|
|
|
|
let running = false;
|
|
let pending = null;
|
|
let schedulerTimer = null;
|
|
|
|
function jobPriority(type) {
|
|
if (type === 'full') return 3;
|
|
if (type === 'incremental') return 2;
|
|
return 1; // remove
|
|
}
|
|
|
|
function createPending(job) {
|
|
const deferred = {};
|
|
deferred.promise = new Promise((resolve, reject) => {
|
|
deferred.resolve = resolve;
|
|
deferred.reject = reject;
|
|
});
|
|
return {
|
|
type: job.type,
|
|
reason: job.reason,
|
|
removeIds: new Set(job.removeIds || []),
|
|
deferred,
|
|
};
|
|
}
|
|
|
|
function mergePending(existing, incoming) {
|
|
const merged = existing;
|
|
if (jobPriority(incoming.type) > jobPriority(merged.type)) {
|
|
merged.type = incoming.type;
|
|
if (merged.type === 'full') merged.removeIds = new Set();
|
|
}
|
|
if (incoming.reason) merged.reason = incoming.reason;
|
|
if (incoming.removeIds) {
|
|
for (const id of incoming.removeIds) merged.removeIds.add(id);
|
|
}
|
|
return merged;
|
|
}
|
|
|
|
async function executeJob(job) {
|
|
stmts.setInProgress.run(1);
|
|
try {
|
|
if (job.type === 'full') return await runFullReconcile(job.reason);
|
|
if (job.type === 'incremental') {
|
|
if (job.removeIds && job.removeIds.size) {
|
|
for (const id of job.removeIds) removeTaskFromCache(id, `${job.reason}:queued-remove`);
|
|
}
|
|
return await runIncrementalScan(job.reason);
|
|
}
|
|
if (job.type === 'remove') {
|
|
let lastResult = null;
|
|
for (const id of job.removeIds || []) {
|
|
lastResult = removeTaskFromCache(id, job.reason);
|
|
}
|
|
return lastResult || markSuccessNoChange({ fullReconcile: false });
|
|
}
|
|
return markSuccessNoChange({ fullReconcile: false });
|
|
} catch (error) {
|
|
markError(error);
|
|
throw error;
|
|
} finally {
|
|
stmts.setInProgress.run(0);
|
|
}
|
|
}
|
|
|
|
function runJobAndDrain(job) {
|
|
running = true;
|
|
return (async () => {
|
|
let result;
|
|
let error;
|
|
try {
|
|
result = await executeJob(job);
|
|
} catch (e) {
|
|
error = e;
|
|
}
|
|
running = false;
|
|
|
|
if (pending) {
|
|
const next = pending;
|
|
pending = null;
|
|
runJobAndDrain({
|
|
type: next.type,
|
|
reason: next.reason,
|
|
removeIds: next.removeIds,
|
|
}).then(next.deferred.resolve, next.deferred.reject);
|
|
}
|
|
|
|
if (error) throw error;
|
|
return result;
|
|
})();
|
|
}
|
|
|
|
function requestJob(job, { wait = true } = {}) {
|
|
const normalized = {
|
|
type: job.type,
|
|
reason: job.reason || job.type,
|
|
removeIds: new Set(job.removeIds || []),
|
|
};
|
|
|
|
if (!running) {
|
|
const p = runJobAndDrain(normalized);
|
|
if (!wait) p.catch((e) => logger.error('[Refresh] background error:', e.message));
|
|
return wait ? p : Promise.resolve({ queued: false });
|
|
}
|
|
|
|
if (!pending) {
|
|
pending = createPending(normalized);
|
|
} else {
|
|
pending = mergePending(pending, normalized);
|
|
}
|
|
|
|
return wait ? pending.deferred.promise : Promise.resolve({ queued: true });
|
|
}
|
|
|
|
function requestIncremental(reason, opts) {
|
|
return requestJob({ type: 'incremental', reason }, opts);
|
|
}
|
|
|
|
function requestFull(reason, opts) {
|
|
return requestJob({ type: 'full', reason }, opts);
|
|
}
|
|
|
|
function requestRemoveTask(taskId, reason, opts) {
|
|
return requestJob({ type: 'remove', reason, removeIds: [String(taskId)] }, opts);
|
|
}
|
|
|
|
function parseWebhookAction(eventTypeRaw) {
|
|
const eventType = String(eventTypeRaw || '').toLowerCase().trim();
|
|
if (!eventType) return { action: 'ignore', eventType };
|
|
if (CHANGE_EVENT_TYPES.has(eventType)) return { action: 'incremental', eventType };
|
|
if (REMOVE_EVENT_TYPES.has(eventType)) return { action: 'remove', eventType };
|
|
return { action: 'ignore', eventType };
|
|
}
|
|
|
|
function extractWebhookTaskId(body = {}) {
|
|
return (
|
|
body.task_id
|
|
|| body.taskId
|
|
|| body.task?.id
|
|
|| body.data?.task_id
|
|
|| body.data?.taskId
|
|
|| body.data?.task?.id
|
|
|| null
|
|
);
|
|
}
|
|
|
|
function getMode() {
|
|
return normalizeSettings().refresh_mode;
|
|
}
|
|
|
|
function getStatusSnapshot() {
|
|
const sync = getSyncState();
|
|
return {
|
|
cache_version: Number(sync.cache_version || 0),
|
|
refresh_in_progress: !!sync.refresh_in_progress,
|
|
last_success_at: sync.last_success_at || null,
|
|
last_error: sync.last_error || null,
|
|
last_full_reconcile_at: sync.last_full_reconcile_at || null,
|
|
};
|
|
}
|
|
|
|
async function schedulerTick() {
|
|
const settings = normalizeSettings();
|
|
const state = getSyncState();
|
|
|
|
if (shouldRunDailyFull(settings, state)) {
|
|
requestFull('daily_full_reconcile', { wait: false });
|
|
}
|
|
|
|
if (settings.refresh_mode === 'incremental_scan') {
|
|
requestIncremental('scheduler_incremental_scan', { wait: false });
|
|
}
|
|
}
|
|
|
|
function startScheduler() {
|
|
if (schedulerTimer) clearInterval(schedulerTimer);
|
|
schedulerTimer = setInterval(() => {
|
|
schedulerTick().catch((e) => logger.error('[Refresh] scheduler error:', e.message));
|
|
}, SCHEDULER_TICK_MS);
|
|
schedulerTick().catch((e) => logger.error('[Refresh] scheduler bootstrap error:', e.message));
|
|
}
|
|
|
|
function stopScheduler() {
|
|
if (schedulerTimer) {
|
|
clearInterval(schedulerTimer);
|
|
schedulerTimer = null;
|
|
}
|
|
}
|
|
|
|
return {
|
|
readCachedTasks,
|
|
getSyncState,
|
|
getStatusSnapshot,
|
|
getMode,
|
|
normalizeSettings,
|
|
startScheduler,
|
|
stopScheduler,
|
|
requestIncremental,
|
|
requestFull,
|
|
requestRemoveTask,
|
|
parseWebhookAction,
|
|
extractWebhookTaskId,
|
|
};
|
|
}
|
|
|
|
module.exports = {
|
|
createRefreshService,
|
|
};
|