|
|
|
|
@ -10,10 +10,8 @@ const sourceIdService = require('./source_id');
|
|
|
|
|
const dateUtils = require('./date_utils');
|
|
|
|
|
const syncUpdateService = require('./sync_update');
|
|
|
|
|
const contentHashService = require('./content_hash');
|
|
|
|
|
const eventLogService = require('./event_log');
|
|
|
|
|
const fs = require('fs');
|
|
|
|
|
const appInfo = require('./app_info');
|
|
|
|
|
const messagingService = require('./messaging');
|
|
|
|
|
const syncSetup = require('./sync_setup');
|
|
|
|
|
const syncMutexService = require('./sync_mutex');
|
|
|
|
|
const cls = require('./cls');
|
|
|
|
|
@ -91,69 +89,19 @@ async function login() {
|
|
|
|
|
return syncContext;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async function getLastSyncedPull() {
|
|
|
|
|
return parseInt(await optionService.getOption('lastSyncedPull'));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async function setLastSyncedPull(syncId) {
|
|
|
|
|
await optionService.setOption('lastSyncedPull', syncId);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async function pullSync(syncContext) {
|
|
|
|
|
const lastSyncedPull = await getLastSyncedPull();
|
|
|
|
|
const changesUri = '/api/sync/changed?lastSyncId=' + await getLastSyncedPull();
|
|
|
|
|
|
|
|
|
|
const changesUri = '/api/sync/changed?lastSyncId=' + lastSyncedPull;
|
|
|
|
|
const rows = await syncRequest(syncContext, 'GET', changesUri);
|
|
|
|
|
|
|
|
|
|
const syncRows = await syncRequest(syncContext, 'GET', changesUri);
|
|
|
|
|
log.info("Pulled " + rows.length + " changes from " + changesUri);
|
|
|
|
|
|
|
|
|
|
log.info("Pulled " + syncRows.length + " changes from " + changesUri);
|
|
|
|
|
|
|
|
|
|
for (const sync of syncRows) {
|
|
|
|
|
for (const {sync, entity} of rows) {
|
|
|
|
|
if (sourceIdService.isLocalSourceId(sync.sourceId)) {
|
|
|
|
|
log.info(`Skipping pull #${sync.id} ${sync.entityName} ${sync.entityId} because ${sync.sourceId} is a local source id.`);
|
|
|
|
|
|
|
|
|
|
await setLastSyncedPull(sync.id);
|
|
|
|
|
|
|
|
|
|
continue;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
const resp = await syncRequest(syncContext, 'GET', "/api/sync/" + sync.entityName + "/" + encodeURIComponent(sync.entityId));
|
|
|
|
|
|
|
|
|
|
if (!resp || (sync.entityName === 'notes' && !resp.entity)) {
|
|
|
|
|
log.error(`Empty response to pull for sync #${sync.id} ${sync.entityName}, id=${sync.entityId}`);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'notes') {
|
|
|
|
|
await syncUpdateService.updateNote(resp.entity, syncContext.sourceId);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'branches') {
|
|
|
|
|
await syncUpdateService.updateBranch(resp, syncContext.sourceId);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'note_revisions') {
|
|
|
|
|
await syncUpdateService.updateNoteRevision(resp, syncContext.sourceId);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'note_reordering') {
|
|
|
|
|
await syncUpdateService.updateNoteReordering(resp, syncContext.sourceId);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'options') {
|
|
|
|
|
await syncUpdateService.updateOptions(resp, syncContext.sourceId);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'recent_notes') {
|
|
|
|
|
await syncUpdateService.updateRecentNotes(resp, syncContext.sourceId);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'images') {
|
|
|
|
|
await syncUpdateService.updateImage(resp, syncContext.sourceId);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'note_images') {
|
|
|
|
|
await syncUpdateService.updateNoteImage(resp, syncContext.sourceId);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'labels') {
|
|
|
|
|
await syncUpdateService.updateLabel(resp, syncContext.sourceId);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'api_tokens') {
|
|
|
|
|
await syncUpdateService.updateApiToken(resp, syncContext.sourceId);
|
|
|
|
|
}
|
|
|
|
|
else {
|
|
|
|
|
throw new Error(`Unrecognized entity type ${sync.entityName} in sync #${sync.id}`);
|
|
|
|
|
await syncUpdateService.updateEntity(sync.entityName, entity, syncContext.sourceId);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
await setLastSyncedPull(sync.id);
|
|
|
|
|
@ -162,145 +110,69 @@ async function pullSync(syncContext) {
|
|
|
|
|
log.info("Finished pull");
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async function getLastSyncedPush() {
|
|
|
|
|
return parseInt(await optionService.getOption('lastSyncedPush'));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async function setLastSyncedPush(lastSyncedPush) {
|
|
|
|
|
await optionService.setOption('lastSyncedPush', lastSyncedPush);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async function pushSync(syncContext) {
|
|
|
|
|
let lastSyncedPush = await getLastSyncedPush();
|
|
|
|
|
|
|
|
|
|
while (true) {
|
|
|
|
|
const sync = await sql.getRowOrNull('SELECT * FROM sync WHERE id > ? LIMIT 1', [lastSyncedPush]);
|
|
|
|
|
|
|
|
|
|
if (sync === null) {
|
|
|
|
|
// nothing to sync
|
|
|
|
|
|
|
|
|
|
log.info("Nothing to push");
|
|
|
|
|
const syncs = await sql.getRows('SELECT * FROM sync WHERE id > ? LIMIT 1000', [lastSyncedPush]);
|
|
|
|
|
|
|
|
|
|
break;
|
|
|
|
|
}
|
|
|
|
|
const filteredSyncs = syncs.filter(sync => {
|
|
|
|
|
if (sync.sourceId === syncContext.sourceId) {
|
|
|
|
|
log.info(`Skipping push #${sync.id} ${sync.entityName} ${sync.entityId} because it originates from sync target`);
|
|
|
|
|
|
|
|
|
|
if (sync.sourceId === syncContext.sourceId) {
|
|
|
|
|
log.info(`Skipping push #${sync.id} ${sync.entityName} ${sync.entityId} because it originates from sync target`);
|
|
|
|
|
}
|
|
|
|
|
else {
|
|
|
|
|
await pushEntity(sync, syncContext);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
lastSyncedPush = sync.id;
|
|
|
|
|
|
|
|
|
|
await setLastSyncedPush(lastSyncedPush);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
// this may set lastSyncedPush beyond what's actually sent (because of size limit)
|
|
|
|
|
// so this is applied to the database only if there's no actual update
|
|
|
|
|
// TODO: it would be better to simplify this somehow
|
|
|
|
|
lastSyncedPush = sync.id;
|
|
|
|
|
|
|
|
|
|
async function pushEntity(sync, syncContext) {
|
|
|
|
|
let entity;
|
|
|
|
|
return false;
|
|
|
|
|
}
|
|
|
|
|
else {
|
|
|
|
|
return true;
|
|
|
|
|
}
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
if (sync.entityName === 'notes') {
|
|
|
|
|
entity = await sql.getRow('SELECT * FROM notes WHERE noteId = ?', [sync.entityId]);
|
|
|
|
|
if (filteredSyncs.length === 0) {
|
|
|
|
|
log.info("Nothing to push");
|
|
|
|
|
|
|
|
|
|
serializeNoteContentBuffer(entity);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'branches') {
|
|
|
|
|
entity = await sql.getRow('SELECT * FROM branches WHERE branchId = ?', [sync.entityId]);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'note_revisions') {
|
|
|
|
|
entity = await sql.getRow('SELECT * FROM note_revisions WHERE noteRevisionId = ?', [sync.entityId]);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'note_reordering') {
|
|
|
|
|
entity = {
|
|
|
|
|
parentNoteId: sync.entityId,
|
|
|
|
|
ordering: await sql.getMap('SELECT branchId, notePosition FROM branches WHERE parentNoteId = ? AND isDeleted = 0', [sync.entityId])
|
|
|
|
|
};
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'options') {
|
|
|
|
|
entity = await sql.getRow('SELECT * FROM options WHERE name = ?', [sync.entityId]);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'recent_notes') {
|
|
|
|
|
entity = await sql.getRow('SELECT * FROM recent_notes WHERE branchId = ?', [sync.entityId]);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'images') {
|
|
|
|
|
entity = await sql.getRow('SELECT * FROM images WHERE imageId = ?', [sync.entityId]);
|
|
|
|
|
await setLastSyncedPush(lastSyncedPush);
|
|
|
|
|
|
|
|
|
|
if (entity.data !== null) {
|
|
|
|
|
entity.data = entity.data.toString('base64');
|
|
|
|
|
break;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'note_images') {
|
|
|
|
|
entity = await sql.getRow('SELECT * FROM note_images WHERE noteImageId = ?', [sync.entityId]);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'labels') {
|
|
|
|
|
entity = await sql.getRow('SELECT * FROM labels WHERE labelId = ?', [sync.entityId]);
|
|
|
|
|
}
|
|
|
|
|
else if (sync.entityName === 'api_tokens') {
|
|
|
|
|
entity = await sql.getRow('SELECT * FROM api_tokens WHERE apiTokenId = ?', [sync.entityId]);
|
|
|
|
|
}
|
|
|
|
|
else {
|
|
|
|
|
throw new Error(`Unrecognized entity type ${sync.entityName} in sync #${sync.id}`);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (!entity) {
|
|
|
|
|
log.info(`Sync #${sync.id} entity for ${sync.entityName} ${sync.entityId} doesn't exist. Skipping.`);
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
const syncRecords = await getSyncRecords(filteredSyncs);
|
|
|
|
|
|
|
|
|
|
log.info(`Pushing changes in sync #${sync.id} ${sync.entityName} ${sync.entityId}`);
|
|
|
|
|
log.info(`Pushing ${syncRecords.length} syncs.`);
|
|
|
|
|
|
|
|
|
|
const payload = {
|
|
|
|
|
sourceId: sourceIdService.getCurrentSourceId(),
|
|
|
|
|
entity: entity
|
|
|
|
|
};
|
|
|
|
|
await syncRequest(syncContext, 'PUT', '/api/sync/update', {
|
|
|
|
|
sourceId: sourceIdService.getCurrentSourceId(),
|
|
|
|
|
entities: syncRecords
|
|
|
|
|
});
|
|
|
|
|
|
|
|
|
|
await syncRequest(syncContext, 'PUT', '/api/sync/' + sync.entityName, payload);
|
|
|
|
|
}
|
|
|
|
|
lastSyncedPush = syncRecords[syncRecords.length - 1].sync.id;
|
|
|
|
|
|
|
|
|
|
function serializeNoteContentBuffer(note) {
|
|
|
|
|
if (note.type === 'file') {
|
|
|
|
|
note.content = note.content.toString("binary");
|
|
|
|
|
await setLastSyncedPush(lastSyncedPush);
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async function checkContentHash(syncContext) {
|
|
|
|
|
const resp = await syncRequest(syncContext, 'GET', '/api/sync/check');
|
|
|
|
|
|
|
|
|
|
if (await getLastSyncedPull() < resp.max_sync_id) {
|
|
|
|
|
if (await getLastSyncedPull() < resp.maxSyncId) {
|
|
|
|
|
log.info("There are some outstanding pulls, skipping content check.");
|
|
|
|
|
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
const lastSyncedPush = await getLastSyncedPush();
|
|
|
|
|
const notPushedSyncs = await sql.getValue("SELECT COUNT(*) FROM sync WHERE id > ?", [lastSyncedPush]);
|
|
|
|
|
const notPushedSyncs = await sql.getValue("SELECT COUNT(*) FROM sync WHERE id > ?", [await getLastSyncedPush()]);
|
|
|
|
|
|
|
|
|
|
if (notPushedSyncs > 0) {
|
|
|
|
|
log.info("There's " + notPushedSyncs + " outstanding pushes, skipping content check.");
|
|
|
|
|
log.info(`There's ${notPushedSyncs} outstanding pushes, skipping content check.`);
|
|
|
|
|
|
|
|
|
|
return;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
const hashes = await contentHashService.getHashes();
|
|
|
|
|
let allChecksPassed = true;
|
|
|
|
|
|
|
|
|
|
for (const key in hashes) {
|
|
|
|
|
if (hashes[key] !== resp.hashes[key]) {
|
|
|
|
|
allChecksPassed = false;
|
|
|
|
|
|
|
|
|
|
await eventLogService.addEvent(`Content hash check for ${key} FAILED. Local is ${hashes[key]}, remote is ${resp.hashes[key]}`);
|
|
|
|
|
|
|
|
|
|
if (key !== 'recent_notes') {
|
|
|
|
|
// let's not get alarmed about recent notes which get updated often and can cause failures in race conditions
|
|
|
|
|
await messagingService.sendMessageToAllClients({type: 'sync-hash-check-failed'});
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
if (allChecksPassed) {
|
|
|
|
|
log.info("Content hash checks PASSED");
|
|
|
|
|
}
|
|
|
|
|
await contentHashService.checkContentHashes(resp.hashes);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async function syncRequest(syncContext, method, uri, body) {
|
|
|
|
|
@ -331,6 +203,80 @@ async function syncRequest(syncContext, method, uri, body) {
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
const primaryKeys = {
|
|
|
|
|
"notes": "noteId",
|
|
|
|
|
"branches": "branchId",
|
|
|
|
|
"note_revisions": "noteRevisionId",
|
|
|
|
|
"option": "name",
|
|
|
|
|
"recent_notes": "branchId",
|
|
|
|
|
"images": "imageId",
|
|
|
|
|
"note_images": "noteImageId",
|
|
|
|
|
"labels": "labelId",
|
|
|
|
|
"api_tokens": "apiTokenId"
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
async function getEntityRow(entityName, entityId) {
|
|
|
|
|
if (entityName === 'note_reordering') {
|
|
|
|
|
return await sql.getMap("SELECT branchId, notePosition FROM branches WHERE parentNoteId = ? AND isDeleted = 0", [entityId]);
|
|
|
|
|
}
|
|
|
|
|
else {
|
|
|
|
|
const primaryKey = primaryKeys[entityName];
|
|
|
|
|
|
|
|
|
|
if (!primaryKey) {
|
|
|
|
|
throw new Error("Unknown entity " + entityName);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
const entity = await sql.getRow(`SELECT * FROM ${entityName} WHERE ${primaryKey} = ?`, [entityId]);
|
|
|
|
|
|
|
|
|
|
if (entityName === 'notes' && entity.type === 'file') {
|
|
|
|
|
entity.content = entity.content.toString("binary");
|
|
|
|
|
}
|
|
|
|
|
else if (entityName === 'images') {
|
|
|
|
|
entity.data = entity.data.toString('base64');
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return entity;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async function getSyncRecords(syncs) {
|
|
|
|
|
const records = [];
|
|
|
|
|
let length = 0;
|
|
|
|
|
|
|
|
|
|
for (const sync of syncs) {
|
|
|
|
|
const record = {
|
|
|
|
|
sync: sync,
|
|
|
|
|
entity: await getEntityRow(sync.entityName, sync.entityId)
|
|
|
|
|
};
|
|
|
|
|
|
|
|
|
|
records.push(record);
|
|
|
|
|
|
|
|
|
|
length += JSON.stringify(record).length;
|
|
|
|
|
|
|
|
|
|
if (length > 1000000) {
|
|
|
|
|
break;
|
|
|
|
|
}
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
return records;
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async function getLastSyncedPull() {
|
|
|
|
|
return parseInt(await optionService.getOption('lastSyncedPull'));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async function setLastSyncedPull(syncId) {
|
|
|
|
|
await optionService.setOption('lastSyncedPull', syncId);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async function getLastSyncedPush() {
|
|
|
|
|
return parseInt(await optionService.getOption('lastSyncedPush'));
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
async function setLastSyncedPush(lastSyncedPush) {
|
|
|
|
|
await optionService.setOption('lastSyncedPush', lastSyncedPush);
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
sqlInit.dbReady.then(() => {
|
|
|
|
|
if (syncSetup.isSyncSetup) {
|
|
|
|
|
log.info("Setting up sync to " + syncSetup.SYNC_SERVER + " with timeout " + syncSetup.SYNC_TIMEOUT);
|
|
|
|
|
@ -357,5 +303,5 @@ sqlInit.dbReady.then(() => {
|
|
|
|
|
|
|
|
|
|
module.exports = {
|
|
|
|
|
sync,
|
|
|
|
|
serializeNoteContentBuffer
|
|
|
|
|
getSyncRecords
|
|
|
|
|
};
|
azivner