915 lines
30 KiB
JavaScript
915 lines
30 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',
|
|
]);
|
|
|
|
const COMMENT_EVENT_TYPES = new Set([50, 55]);
|
|
|
|
const INTRA_ADDFIELD_ALIAS = 'addfield_2f6eaa2vnutrennij_statusSingleSelect';
|
|
|
|
function toBoolFlag(value) {
|
|
return String(value || '0') === '1' || value === true;
|
|
}
|
|
|
|
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 client_dict (
|
|
client_id INTEGER PRIMARY KEY,
|
|
name TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS client_users (
|
|
user_id INTEGER NOT NULL,
|
|
name TEXT NOT NULL,
|
|
group_id INTEGER NOT NULL,
|
|
PRIMARY KEY (user_id, group_id)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_client_users_group ON client_users(group_id);
|
|
|
|
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(?,?,?)'),
|
|
upsertClient: db.prepare(`
|
|
INSERT INTO client_dict(client_id, name, updated_at) VALUES(?,?,?)
|
|
ON CONFLICT(client_id) DO UPDATE SET name=excluded.name, updated_at=excluded.updated_at
|
|
`),
|
|
clearClientUsers: db.prepare('DELETE FROM client_users'),
|
|
upsertClientUser: db.prepare('INSERT OR REPLACE INTO client_users(user_id, name, group_id) VALUES(?,?,?)'),
|
|
upsertIntraStatus: db.prepare(`
|
|
INSERT INTO intra_status_dict(status_id,name,updated_at) VALUES(?,?,?)
|
|
ON CONFLICT(status_id) DO UPDATE SET name=excluded.name,updated_at=excluded.updated_at
|
|
`),
|
|
getSystemStatuses: db.prepare(`
|
|
SELECT key, system_role
|
|
FROM internal_statuses
|
|
WHERE is_active=1 AND system_role IS NOT NULL
|
|
ORDER BY sort_order ASC, key ASC
|
|
`),
|
|
getTaskActiveMark: db.prepare(`
|
|
SELECT mark_type
|
|
FROM marks
|
|
WHERE task_id=? AND is_active=1
|
|
ORDER BY set_at DESC, mark_type ASC
|
|
LIMIT 1
|
|
`),
|
|
clearTaskMarks: db.prepare('UPDATE marks SET is_active=0,set_by=?,set_at=? WHERE task_id=? AND is_active=1'),
|
|
clearOtherTaskMarks: db.prepare('UPDATE marks SET is_active=0,set_by=?,set_at=? WHERE task_id=? AND mark_type<>? AND is_active=1'),
|
|
upsertTaskMark: db.prepare(`
|
|
INSERT INTO marks(task_id,mark_type,is_active,set_by,set_at) VALUES(?,?,?,?,?)
|
|
ON CONFLICT(task_id,mark_type) DO UPDATE SET is_active=excluded.is_active,set_by=excluded.set_by,set_at=excluded.set_at
|
|
`),
|
|
getStatusKeyByLabel: db.prepare(
|
|
"SELECT key FROM internal_statuses WHERE is_active=1 AND lower(label)=lower(?) LIMIT 1"
|
|
),
|
|
};
|
|
|
|
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();
|
|
const doneIds = String(s.intra_done_status_ids || '')
|
|
.split(',')
|
|
.map((v) => String(v).trim())
|
|
.filter(Boolean);
|
|
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),
|
|
auto_reset_on_update: toBoolFlag(s.auto_reset_on_update),
|
|
sync_reply_wait: toBoolFlag(s.sync_reply_wait),
|
|
sync_done_with_intra: toBoolFlag(s.sync_done_with_intra),
|
|
sync_addfield_status: toBoolFlag(s.sync_addfield_status),
|
|
intra_done_status_ids: doneIds,
|
|
};
|
|
}
|
|
|
|
function upsertIntraStatusesFromTasks(tasks) {
|
|
const now = safeIsoNow();
|
|
const statusMap = new Map();
|
|
for (const task of tasks || []) {
|
|
const statusId = String(task?.statusid || task?.status || '').trim();
|
|
const statusName = String(task?._statusname || task?.statusname || '').trim();
|
|
if (!statusId || !statusName) continue;
|
|
statusMap.set(statusId, statusName);
|
|
}
|
|
if (!statusMap.size) return;
|
|
const tx = db.transaction(() => {
|
|
for (const [statusId, name] of statusMap.entries()) {
|
|
stmts.upsertIntraStatus.run(statusId, name, now);
|
|
}
|
|
});
|
|
tx();
|
|
}
|
|
|
|
function getSystemStatusKeys() {
|
|
const rows = stmts.getSystemStatuses.all();
|
|
const roles = {};
|
|
for (const row of rows) {
|
|
const role = String(row.system_role || '').trim();
|
|
if (!role) continue;
|
|
if (!roles[role]) roles[role] = String(row.key);
|
|
}
|
|
return roles;
|
|
}
|
|
|
|
function getCurrentActiveMark(taskId) {
|
|
return stmts.getTaskActiveMark.get(String(taskId))?.mark_type || null;
|
|
}
|
|
|
|
function setExclusiveMark(taskId, markType, setBy = 'system') {
|
|
const now = safeIsoNow();
|
|
const id = String(taskId);
|
|
stmts.clearOtherTaskMarks.run(setBy, now, id, markType);
|
|
stmts.upsertTaskMark.run(id, markType, 1, setBy, now);
|
|
}
|
|
|
|
function clearTaskMarks(taskId, setBy = 'system') {
|
|
const now = safeIsoNow();
|
|
stmts.clearTaskMarks.run(setBy, now, String(taskId));
|
|
}
|
|
|
|
async function fetchLatestCommentAuthorType(taskId, apiKey) {
|
|
if (!apiKey) return null;
|
|
try {
|
|
const url = `https://apigw.intradesk.ru/taskhistory/api/v2.0/lifetime/${taskId}/full`
|
|
+ `?ApiKey=${apiKey}&sortDirection=Desc&top=30&events=50&events=55`;
|
|
const response = await fetch(url, { timeout: 15000 });
|
|
if (!response.ok) return null;
|
|
const json = await response.json();
|
|
const entries = json?.data || [];
|
|
for (const entry of entries) {
|
|
const events = entry?.events?.data || [];
|
|
const hasComment = events.some((ev) => COMMENT_EVENT_TYPES.has(ev?.type));
|
|
if (!hasComment) continue;
|
|
if (entry?.usertype === 10) return 'internal';
|
|
if (entry?.usertype === 20) return 'external';
|
|
return null;
|
|
}
|
|
return null;
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
async function updateIntraAdditionalField(taskNumber, statusLabel, apiKey) {
|
|
if (!apiKey) throw new Error('API ключ не задан');
|
|
const url = `https://apigw.intradesk.ru/changes/v3/tasks/?ApiKey=${apiKey}`;
|
|
const response = await fetch(url, {
|
|
method: 'PUT',
|
|
headers: { 'Content-Type': 'application/json' },
|
|
body: JSON.stringify({
|
|
number: Number(taskNumber),
|
|
blocks: { [INTRA_ADDFIELD_ALIAS]: JSON.stringify({ value: statusLabel || null }) },
|
|
Channel: 'api',
|
|
}),
|
|
timeout: 15000,
|
|
});
|
|
if (!response.ok) {
|
|
const text = await response.text().catch(() => '');
|
|
throw new Error(`IntraDesk вернул ${response.status}: ${text.slice(0, 200)}`);
|
|
}
|
|
}
|
|
|
|
async function applyStatusAutomationForChangedTasks(changedTaskIds, tasksById, settings, reason = '') {
|
|
const changedIds = [...(changedTaskIds || [])].map((v) => String(v));
|
|
if (!changedIds.length) return;
|
|
if (!settings.sync_done_with_intra && !settings.sync_reply_wait && !settings.auto_reset_on_update && !settings.sync_addfield_status) return;
|
|
|
|
const systemStatuses = getSystemStatusKeys();
|
|
const doneKey = systemStatuses.done || null;
|
|
const needsReplyKey = systemStatuses.needs_reply || null;
|
|
const waitClientKey = systemStatuses.wait_client || null;
|
|
const doneIds = new Set((settings.intra_done_status_ids || []).map((v) => String(v)));
|
|
|
|
for (const taskId of changedIds) {
|
|
const task = tasksById.get(taskId);
|
|
if (!task) continue;
|
|
let currentMark = getCurrentActiveMark(taskId);
|
|
let applied = false;
|
|
|
|
if (settings.sync_done_with_intra && doneKey && doneIds.size) {
|
|
const statusId = String(task?.statusid || task?.status || '').trim();
|
|
if (statusId && doneIds.has(statusId)) {
|
|
if (currentMark !== doneKey) {
|
|
setExclusiveMark(taskId, doneKey, 'system');
|
|
}
|
|
applied = true;
|
|
currentMark = doneKey;
|
|
}
|
|
}
|
|
|
|
if (!applied && settings.sync_reply_wait && currentMark && needsReplyKey && waitClientKey) {
|
|
const inPair = currentMark === needsReplyKey || currentMark === waitClientKey;
|
|
if (inPair) {
|
|
const authorType = await fetchLatestCommentAuthorType(taskId, settings.api_key);
|
|
const target = authorType === 'internal'
|
|
? waitClientKey
|
|
: (authorType === 'external' ? needsReplyKey : null);
|
|
if (target) {
|
|
if (currentMark !== target) {
|
|
setExclusiveMark(taskId, target, 'system');
|
|
}
|
|
applied = true;
|
|
currentMark = target;
|
|
} else {
|
|
logger.log(`[Refresh] Pair sync skipped: no author type for task ${taskId} (${reason})`);
|
|
}
|
|
}
|
|
}
|
|
|
|
if (!applied && settings.auto_reset_on_update && currentMark) {
|
|
clearTaskMarks(taskId, 'system');
|
|
applied = true;
|
|
}
|
|
|
|
if (settings.sync_addfield_status) {
|
|
const addfieldVal = task._addfield_status;
|
|
if (addfieldVal) {
|
|
const row = stmts.getStatusKeyByLabel.get(addfieldVal);
|
|
if (row && currentMark !== row.key) {
|
|
setExclusiveMark(taskId, row.key, 'system');
|
|
logger.log(`[Refresh] Inbound addfield sync: task ${taskId} → mark "${row.key}" (${reason})`);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
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; });
|
|
|
|
// Сохраняем клиентов (usergroup) в отдельную таблицу
|
|
const now = safeIsoNow();
|
|
const clientEntries = (json.dictionaries || []).filter((d) => d.type === 'usergroup' && !d.isarchived);
|
|
const upsertClients = db.transaction((entries) => {
|
|
for (const d of entries) stmts.upsertClient.run(d.id, d.name, now);
|
|
});
|
|
if (clientEntries.length) upsertClients(clientEntries);
|
|
|
|
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];
|
|
const af = (task.additionalfields?.data || []).find((d) => d.alias === INTRA_ADDFIELD_ALIAS);
|
|
task._addfield_status = af?.value?.stringvalue || null;
|
|
});
|
|
|
|
upsertIntraStatusesFromTasks(tasks);
|
|
|
|
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]));
|
|
const changedTaskIds = new Set();
|
|
|
|
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 (taskId) changedTaskIds.add(taskId);
|
|
}
|
|
}
|
|
|
|
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})`);
|
|
await applyStatusAutomationForChangedTasks(changedTaskIds, byId, settings, 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;
|
|
const currentById = new Map(currentTasks.map((t) => [extractTaskId(t), extractUpdatedAt(t)]));
|
|
const nextById = new Map(nextTasks.map((t) => [extractTaskId(t), t]));
|
|
const changedTaskIds = new Set();
|
|
for (const task of nextTasks) {
|
|
const taskId = extractTaskId(task);
|
|
if (!taskId) continue;
|
|
const prevUpdated = currentById.get(taskId);
|
|
if (prevUpdated !== extractUpdatedAt(task)) changedTaskIds.add(taskId);
|
|
}
|
|
|
|
if (!hasChangesByIdAndUpdated(currentTasks, nextTasks)) {
|
|
logger.log(`[Refresh] Full reconcile: no changes (${reason})`);
|
|
return markSuccessNoChange({ fullReconcile: true });
|
|
}
|
|
|
|
logger.log(`[Refresh] Full reconcile: changed (${reason})`);
|
|
await applyStatusAutomationForChangedTasks(changedTaskIds, nextById, settings, 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;
|
|
}
|
|
|
|
function firstDefined(obj, keys) {
|
|
for (const key of keys) {
|
|
const value = obj?.[key];
|
|
if (value !== undefined && value !== null && String(value).trim() !== '') return value;
|
|
}
|
|
return null;
|
|
}
|
|
|
|
function extractClientUserRecord(raw) {
|
|
const userId = firstDefined(raw, ['userId', 'userid', 'user_id', 'id']);
|
|
const name = String(firstDefined(raw, ['name', 'fullName', 'fullname', 'displayName', 'username']) || '').trim();
|
|
|
|
const groupIds = new Set();
|
|
const pushGroup = (value) => {
|
|
if (value === undefined || value === null) return;
|
|
const normalized = String(value).trim();
|
|
if (normalized) groupIds.add(normalized);
|
|
};
|
|
|
|
pushGroup(firstDefined(raw, ['userGroupId', 'usergroupid', 'user_group_id', 'groupId', 'group_id']));
|
|
|
|
for (const bucketName of ['groups', 'userGroups']) {
|
|
const bucket = raw?.[bucketName];
|
|
if (!Array.isArray(bucket)) continue;
|
|
for (const group of bucket) {
|
|
pushGroup(firstDefined(group, ['userGroupId', 'usergroupid', 'groupId', 'group_id', 'id']));
|
|
}
|
|
}
|
|
|
|
return {
|
|
userId: userId === null ? '' : String(userId).trim(),
|
|
name,
|
|
groupIds: [...groupIds],
|
|
};
|
|
}
|
|
|
|
async function syncClientUsers() {
|
|
const s = getSettings();
|
|
if (!s.api_key) return { fetched: 0, stored: 0, odata_count: null };
|
|
const pageSize = 400;
|
|
let skip = 0;
|
|
let odataCount = null;
|
|
const sourceRows = [];
|
|
while (true) {
|
|
const url = `https://apigw.intradesk.ru/settings/odata/ClientUsers?ApiKey=${s.api_key}`
|
|
+ `&$count=true`
|
|
+ `&$orderby=${encodeURIComponent('id desc')}`
|
|
+ `&$filter=${encodeURIComponent('(isarchived eq false)')}`
|
|
+ `&$top=${pageSize}&$skip=${skip}`;
|
|
const response = await fetch(url, { timeout: 30000 });
|
|
if (!response.ok) {
|
|
const body = await response.text().catch(() => '');
|
|
throw new Error(`ClientUsers request failed (${response.status}): ${body.slice(0, 180)}`);
|
|
}
|
|
const json = await response.json();
|
|
const rows = Array.isArray(json?.value) ? json.value : [];
|
|
if (odataCount === null && Number.isFinite(Number(json?.['@odata.count']))) {
|
|
odataCount = Number(json['@odata.count']);
|
|
}
|
|
sourceRows.push(...rows);
|
|
if (rows.length < pageSize) break;
|
|
skip += pageSize;
|
|
}
|
|
|
|
const memberships = [];
|
|
const seenMemberships = new Set();
|
|
for (const raw of sourceRows) {
|
|
const parsed = extractClientUserRecord(raw || {});
|
|
if (!parsed.userId || !parsed.name || !parsed.groupIds.length) continue;
|
|
for (const groupId of parsed.groupIds) {
|
|
const dedupeKey = `${parsed.userId}|${groupId}`;
|
|
if (seenMemberships.has(dedupeKey)) continue;
|
|
seenMemberships.add(dedupeKey);
|
|
memberships.push({ userId: parsed.userId, name: parsed.name, groupId });
|
|
}
|
|
}
|
|
|
|
const tx = db.transaction(() => {
|
|
stmts.clearClientUsers.run();
|
|
for (const row of memberships) {
|
|
stmts.upsertClientUser.run(row.userId, row.name, row.groupId);
|
|
}
|
|
});
|
|
tx();
|
|
logger.log(`[Refresh] syncClientUsers: fetched ${sourceRows.length}, stored ${memberships.length}${odataCount !== null ? `, odata.count=${odataCount}` : ''}`);
|
|
return { fetched: sourceRows.length, stored: memberships.length, odata_count: odataCount };
|
|
}
|
|
|
|
let running = false;
|
|
let pending = null;
|
|
let schedulerTimer = null;
|
|
let lastClientUsersSync = 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 });
|
|
}
|
|
|
|
const now = new Date();
|
|
const needUsersSync = !lastClientUsersSync
|
|
|| (now - lastClientUsersSync) > 24 * 60 * 60 * 1000;
|
|
if (needUsersSync) {
|
|
lastClientUsersSync = now;
|
|
syncClientUsers().catch((e) => logger.error('[Refresh] syncClientUsers error:', e.message));
|
|
}
|
|
}
|
|
|
|
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,
|
|
updateIntraAdditionalField,
|
|
syncClientUsers,
|
|
};
|
|
}
|
|
|
|
module.exports = {
|
|
createRefreshService,
|
|
};
|