Dockery/Trilium
1
0
Fork 0
mirror of https://github.com/zadam/trilium.git synced 2025-11-03 20:06:08 +01:00
Code Issues Packages Projects Releases Activity

server-ts: Port services/sync

Browse Source
This commit is contained in:
Elian Doran
2024-02-18 13:10:51 +02:00
parent 43c89c0e9d
commit d89b791914
11 changed files with 133 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/app.js
View File

@@ -43,7 +43,7 @@ require('./routes/custom.js').register(app);
require('./routes/error_handlers.js').register(app); require('./routes/error_handlers.js').register(app);
// triggers sync timer // triggers sync timer
require('./services/sync.js'); require('./services/sync');
// triggers backup timer // triggers backup timer
require('./services/backup'); require('./services/backup');

2
src/routes/api/script.js
View File

@@ -3,7 +3,7 @@
const scriptService = require('../../services/script.js'); const scriptService = require('../../services/script.js');
const attributeService = require('../../services/attributes'); const attributeService = require('../../services/attributes');
const becca = require('../../becca/becca'); const becca = require('../../becca/becca');
const syncService = require('../../services/sync.js'); const syncService = require('../../services/sync');
const sql = require('../../services/sql'); const sql = require('../../services/sql');
// The async/await here is very confusing, because the body.script may, but may not be async. If it is async, then we // The async/await here is very confusing, because the body.script may, but may not be async. If it is async, then we

2
src/routes/api/sync.js
View File

@@ -1,6 +1,6 @@
"use strict"; "use strict";
const syncService = require('../../services/sync.js'); const syncService = require('../../services/sync');
const syncUpdateService = require('../../services/sync_update'); const syncUpdateService = require('../../services/sync_update');
const entityChangesService = require('../../services/entity_changes'); const entityChangesService = require('../../services/entity_changes');
const sql = require('../../services/sql'); const sql = require('../../services/sql');

2
src/routes/routes.js
View File

@@ -32,7 +32,7 @@ const revisionsApiRoute = require('./api/revisions');
const recentChangesApiRoute = require('./api/recent_changes.js'); const recentChangesApiRoute = require('./api/recent_changes.js');
const optionsApiRoute = require('./api/options.js'); const optionsApiRoute = require('./api/options.js');
const passwordApiRoute = require('./api/password'); const passwordApiRoute = require('./api/password');
const syncApiRoute = require('./api/sync.js'); const syncApiRoute = require('./api/sync');
const loginApiRoute = require('./api/login.js'); const loginApiRoute = require('./api/login.js');
const recentNotesRoute = require('./api/recent_notes.js'); const recentNotesRoute = require('./api/recent_notes.js');
const appInfoRoute = require('./api/app_info'); const appInfoRoute = require('./api/app_info');

10
src/services/entity_changes_interface.ts
View File

@@ -13,3 +13,13 @@ export interface EntityChange {
changeId?: string | null; changeId?: string | null;
instanceId?: string | null; instanceId?: string | null;
} }
export interface EntityRow {
isDeleted?: boolean;
content?: Buffer | string;
}
export interface EntityChangeRecord {
entityChange: EntityChange;
entity?: EntityRow;
}

20
src/services/request.ts
View File

@@ -4,29 +4,11 @@ import utils = require('./utils');
import log = require('./log'); import log = require('./log');
import url = require('url'); import url = require('url');
import syncOptions = require('./sync_options'); import syncOptions = require('./sync_options');
import { ExecOpts } from './request_interface';
// this service provides abstraction over node's HTTP/HTTPS and electron net.client APIs // this service provides abstraction over node's HTTP/HTTPS and electron net.client APIs
// this allows supporting system proxy // this allows supporting system proxy
interface ExecOpts {
proxy: "noproxy" | null;
method: string;
url: string;
paging?: {
pageCount: number;
pageIndex: number;
requestId: string;
};
cookieJar?: {
header?: string;
};
auth?: {
password?: string;
},
timeout: number;
body: string;
}
interface ClientOpts { interface ClientOpts {
method: string; method: string;
url: string; url: string;

20
src/services/request_interface.ts Normal file
View File

@@ -0,0 +1,20 @@
export interface CookieJar {
header?: string;
}
export interface ExecOpts {
proxy: "noproxy" | null;
method: string;
url: string;
paging?: {
pageCount: number;
pageIndex: number;
requestId: string;
};
cookieJar?: CookieJar;
auth?: {
password?: string;
},
timeout: number;
body: string;
}

2
src/services/setup.js
View File

@@ -1,4 +1,4 @@
const syncService = require('./sync.js'); const syncService = require('./sync');
const log = require('./log'); const log = require('./log');
const sqlInit = require('./sql_init'); const sqlInit = require('./sql_init');
const optionService = require('./options'); const optionService = require('./options');

130
src/services/sync.js → src/services/sync.ts
View File

@@ -1,27 +1,50 @@
"use strict"; "use strict";
const log = require('./log'); import log = require('./log');
const sql = require('./sql'); import sql = require('./sql');
const optionService = require('./options'); import optionService = require('./options');
const utils = require('./utils'); import utils = require('./utils');
const instanceId = require('./instance_id'); import instanceId = require('./instance_id');
const dateUtils = require('./date_utils'); import dateUtils = require('./date_utils');
const syncUpdateService = require('./sync_update'); import syncUpdateService = require('./sync_update');
const contentHashService = require('./content_hash'); import contentHashService = require('./content_hash');
const appInfo = require('./app_info'); import appInfo = require('./app_info');
const syncOptions = require('./sync_options'); import syncOptions = require('./sync_options');
const syncMutexService = require('./sync_mutex'); import syncMutexService = require('./sync_mutex');
const cls = require('./cls'); import cls = require('./cls');
const request = require('./request'); import request = require('./request');
const ws = require('./ws'); import ws = require('./ws');
const entityChangesService = require('./entity_changes'); import entityChangesService = require('./entity_changes');
const entityConstructor = require('../becca/entity_constructor'); import entityConstructor = require('../becca/entity_constructor');
const becca = require('../becca/becca'); import becca = require('../becca/becca');
import { EntityChange, EntityChangeRecord, EntityRow } from './entity_changes_interface';
import { CookieJar, ExecOpts } from './request_interface';
let proxyToggle = true; let proxyToggle = true;
let outstandingPullCount = 0; let outstandingPullCount = 0;
interface CheckResponse {
maxEntityChangeId: number;
entityHashes: Record<string, Record<string, string>>
}
interface SyncResponse {
instanceId: string;
maxEntityChangeId: number;
}
interface ChangesResponse {
entityChanges: EntityChangeRecord[];
lastEntityChangeId: number;
outstandingPullCount: number;
}
interface SyncContext {
cookieJar: CookieJar;
instanceId?: string;
}
async function sync() { async function sync() {
try { try {
return await syncMutexService.doExclusively(async () => { return await syncMutexService.doExclusively(async () => {
@@ -53,7 +76,7 @@ async function sync() {
}; };
}); });
} }
catch (e) { catch (e: any) {
// we're dynamically switching whether we're using proxy or not based on whether we encountered error with the current method // we're dynamically switching whether we're using proxy or not based on whether we encountered error with the current method
proxyToggle = !proxyToggle; proxyToggle = !proxyToggle;
@@ -93,19 +116,23 @@ async function login() {
return await doLogin(); return await doLogin();
} }
async function doLogin() { async function doLogin(): Promise<SyncContext> {
const timestamp = dateUtils.utcNowDateTime(); const timestamp = dateUtils.utcNowDateTime();
const documentSecret = optionService.getOption('documentSecret'); const documentSecret = optionService.getOption('documentSecret');
const hash = utils.hmac(documentSecret, timestamp); const hash = utils.hmac(documentSecret, timestamp);
const syncContext = { cookieJar: {} }; const syncContext: SyncContext = { cookieJar: {} };
const resp = await syncRequest(syncContext, 'POST', '/api/login/sync', { const resp = await syncRequest<SyncResponse>(syncContext, 'POST', '/api/login/sync', {
timestamp: timestamp, timestamp: timestamp,
syncVersion: appInfo.syncVersion, syncVersion: appInfo.syncVersion,
hash: hash hash: hash
}); });
if (!resp) {
throw new Error("Got no response.");
}
if (resp.instanceId === instanceId) { if (resp.instanceId === instanceId) {
throw new Error(`Sync server has instance ID '${resp.instanceId}' which is also local. This usually happens when the sync client is (mis)configured to sync with itself (URL points back to client) instead of the correct sync server.`); throw new Error(`Sync server has instance ID '${resp.instanceId}' which is also local. This usually happens when the sync client is (mis)configured to sync with itself (URL points back to client) instead of the correct sync server.`);
} }
@@ -125,7 +152,7 @@ async function doLogin() {
return syncContext; return syncContext;
} }
async function pullChanges(syncContext) { async function pullChanges(syncContext: SyncContext) {
while (true) { while (true) {
const lastSyncedPull = getLastSyncedPull(); const lastSyncedPull = getLastSyncedPull();
const logMarkerId = utils.randomString(10); // to easily pair sync events between client and server logs const logMarkerId = utils.randomString(10); // to easily pair sync events between client and server logs
@@ -133,7 +160,10 @@ async function pullChanges(syncContext) {
const startDate = Date.now(); const startDate = Date.now();
const resp = await syncRequest(syncContext, 'GET', changesUri); const resp = await syncRequest<ChangesResponse>(syncContext, 'GET', changesUri);
if (!resp) {
throw new Error("Request failed.");
}
const {entityChanges, lastEntityChangeId} = resp; const {entityChanges, lastEntityChangeId} = resp;
outstandingPullCount = resp.outstandingPullCount; outstandingPullCount = resp.outstandingPullCount;
@@ -141,7 +171,9 @@ async function pullChanges(syncContext) {
const pulledDate = Date.now(); const pulledDate = Date.now();
sql.transactional(() => { sql.transactional(() => {
if (syncContext.instanceId) {
syncUpdateService.updateEntities(entityChanges, syncContext.instanceId); syncUpdateService.updateEntities(entityChanges, syncContext.instanceId);
}
if (lastSyncedPull !== lastEntityChangeId) { if (lastSyncedPull !== lastEntityChangeId) {
setLastSyncedPull(lastEntityChangeId); setLastSyncedPull(lastEntityChangeId);
@@ -156,7 +188,7 @@ async function pullChanges(syncContext) {
log.info(`Sync ${logMarkerId}: Pulled ${entityChanges.length} changes in ${sizeInKb} KB, starting at entityChangeId=${lastSyncedPull} in ${pulledDate - startDate}ms and applied them in ${Date.now() - pulledDate}ms, ${outstandingPullCount} outstanding pulls`); log.info(`Sync ${logMarkerId}: Pulled ${entityChanges.length} changes in ${sizeInKb} KB, starting at entityChangeId=${lastSyncedPull} in ${pulledDate - startDate}ms and applied them in ${Date.now() - pulledDate}ms, ${outstandingPullCount} outstanding pulls`);
} }
catch (e) { catch (e: any) {
log.error(`Error occurred ${e.message} ${e.stack}`); log.error(`Error occurred ${e.message} ${e.stack}`);
} }
} }
@@ -165,11 +197,11 @@ async function pullChanges(syncContext) {
log.info("Finished pull"); log.info("Finished pull");
} }
async function pushChanges(syncContext) { async function pushChanges(syncContext: SyncContext) {
let lastSyncedPush = getLastSyncedPush(); let lastSyncedPush: number | null | undefined = getLastSyncedPush();
while (true) { while (true) {
const entityChanges = sql.getRows('SELECT * FROM entity_changes WHERE isSynced = 1 AND id > ? LIMIT 1000', [lastSyncedPush]); const entityChanges = sql.getRows<EntityChange>('SELECT * FROM entity_changes WHERE isSynced = 1 AND id > ? LIMIT 1000', [lastSyncedPush]);
if (entityChanges.length === 0) { if (entityChanges.length === 0) {
log.info("Nothing to push"); log.info("Nothing to push");
@@ -190,7 +222,7 @@ async function pushChanges(syncContext) {
} }
}); });
if (filteredEntityChanges.length === 0) { if (filteredEntityChanges.length === 0 && lastSyncedPush) {
// there still might be more sync changes (because of batch limit), just all the current batch // there still might be more sync changes (because of batch limit), just all the current batch
// has been filtered out // has been filtered out
setLastSyncedPush(lastSyncedPush); setLastSyncedPush(lastSyncedPush);
@@ -214,16 +246,22 @@ async function pushChanges(syncContext) {
lastSyncedPush = entityChangesRecords[entityChangesRecords.length - 1].entityChange.id; lastSyncedPush = entityChangesRecords[entityChangesRecords.length - 1].entityChange.id;
if (lastSyncedPush) {
setLastSyncedPush(lastSyncedPush); setLastSyncedPush(lastSyncedPush);
} }
}
} }
async function syncFinished(syncContext) { async function syncFinished(syncContext: SyncContext) {
await syncRequest(syncContext, 'POST', '/api/sync/finished'); await syncRequest(syncContext, 'POST', '/api/sync/finished');
} }
async function checkContentHash(syncContext) { async function checkContentHash(syncContext: SyncContext) {
const resp = await syncRequest(syncContext, 'GET', '/api/sync/check'); const resp = await syncRequest<CheckResponse>(syncContext, 'GET', '/api/sync/check');
if (!resp) {
throw new Error("Got no response.");
}
const lastSyncedPullId = getLastSyncedPull(); const lastSyncedPullId = getLastSyncedPull();
if (lastSyncedPullId < resp.maxEntityChangeId) { if (lastSyncedPullId < resp.maxEntityChangeId) {
@@ -261,8 +299,12 @@ async function checkContentHash(syncContext) {
const PAGE_SIZE = 1000000; const PAGE_SIZE = 1000000;
async function syncRequest(syncContext, method, requestPath, body) { interface SyncContext {
body = body ? JSON.stringify(body) : ''; cookieJar: CookieJar
}
async function syncRequest<T extends {}>(syncContext: SyncContext, method: string, requestPath: string, _body?: {}) {
const body = _body ? JSON.stringify(_body) : '';
const timeout = syncOptions.getSyncTimeout(); const timeout = syncOptions.getSyncTimeout();
@@ -272,7 +314,7 @@ async function syncRequest(syncContext, method, requestPath, body) {
const pageCount = Math.max(1, Math.ceil(body.length / PAGE_SIZE)); const pageCount = Math.max(1, Math.ceil(body.length / PAGE_SIZE));
for (let pageIndex = 0; pageIndex < pageCount; pageIndex++) { for (let pageIndex = 0; pageIndex < pageCount; pageIndex++) {
const opts = { const opts: ExecOpts = {
method, method,
url: syncOptions.getSyncServerHost() + requestPath, url: syncOptions.getSyncServerHost() + requestPath,
cookieJar: syncContext.cookieJar, cookieJar: syncContext.cookieJar,
@@ -286,13 +328,13 @@ async function syncRequest(syncContext, method, requestPath, body) {
proxy: proxyToggle ? syncOptions.getSyncProxy() : null proxy: proxyToggle ? syncOptions.getSyncProxy() : null
}; };
response = await utils.timeLimit(request.exec(opts), timeout); response = await utils.timeLimit(request.exec(opts), timeout) as T;
} }
return response; return response;
} }
function getEntityChangeRow(entityChange) { function getEntityChangeRow(entityChange: EntityChange) {
const {entityName, entityId} = entityChange; const {entityName, entityId} = entityChange;
if (entityName === 'note_reordering') { if (entityName === 'note_reordering') {
@@ -305,7 +347,7 @@ function getEntityChangeRow(entityChange) {
throw new Error(`Unknown entity for entity change ${JSON.stringify(entityChange)}`); throw new Error(`Unknown entity for entity change ${JSON.stringify(entityChange)}`);
} }
const entityRow = sql.getRow(`SELECT * FROM ${entityName} WHERE ${primaryKey} = ?`, [entityId]); const entityRow = sql.getRow<EntityRow>(`SELECT * FROM ${entityName} WHERE ${primaryKey} = ?`, [entityId]);
if (!entityRow) { if (!entityRow) {
log.error(`Cannot find entity for entity change ${JSON.stringify(entityChange)}`); log.error(`Cannot find entity for entity change ${JSON.stringify(entityChange)}`);
@@ -317,15 +359,17 @@ function getEntityChangeRow(entityChange) {
entityRow.content = Buffer.from(entityRow.content, 'utf-8'); entityRow.content = Buffer.from(entityRow.content, 'utf-8');
} }
if (entityRow.content) {
entityRow.content = entityRow.content.toString("base64"); entityRow.content = entityRow.content.toString("base64");
} }
}
return entityRow; return entityRow;
} }
} }
function getEntityChangeRecords(entityChanges) { function getEntityChangeRecords(entityChanges: EntityChange[]) {
const records = []; const records: EntityChangeRecord[] = [];
let length = 0; let length = 0;
for (const entityChange of entityChanges) { for (const entityChange of entityChanges) {
@@ -340,7 +384,7 @@ function getEntityChangeRecords(entityChanges) {
continue; continue;
} }
const record = { entityChange, entity }; const record: EntityChangeRecord = { entityChange, entity };
records.push(record); records.push(record);
@@ -359,7 +403,7 @@ function getLastSyncedPull() {
return parseInt(optionService.getOption('lastSyncedPull')); return parseInt(optionService.getOption('lastSyncedPull'));
} }
function setLastSyncedPull(entityChangeId) { function setLastSyncedPull(entityChangeId: number) {
const lastSyncedPullOption = becca.getOption('lastSyncedPull'); const lastSyncedPullOption = becca.getOption('lastSyncedPull');
if (lastSyncedPullOption) { // might be null in initial sync when becca is not loaded if (lastSyncedPullOption) { // might be null in initial sync when becca is not loaded
@@ -378,7 +422,7 @@ function getLastSyncedPush() {
return lastSyncedPush; return lastSyncedPush;
} }
function setLastSyncedPush(entityChangeId) { function setLastSyncedPush(entityChangeId: number) {
ws.setLastSyncedPush(entityChangeId); ws.setLastSyncedPush(entityChangeId);
const lastSyncedPushOption = becca.getOption('lastSyncedPush'); const lastSyncedPushOption = becca.getOption('lastSyncedPush');
@@ -409,7 +453,7 @@ require('../becca/becca_loader').beccaLoaded.then(() => {
getLastSyncedPush(); getLastSyncedPush();
}); });
module.exports = { export = {
sync, sync,
login, login,
getEntityChangeRecords, getEntityChangeRecords,

16
src/services/sync_update.ts
View File

@@ -4,17 +4,7 @@ import entityChangesService = require('./entity_changes');
import eventService = require('./events'); import eventService = require('./events');
import entityConstructor = require('../becca/entity_constructor'); import entityConstructor = require('../becca/entity_constructor');
import ws = require('./ws'); import ws = require('./ws');
import { EntityChange } from './entity_changes_interface'; import { EntityChange, EntityChangeRecord, EntityRow } from './entity_changes_interface';
interface EntityRow {
isDeleted?: boolean;
content: Buffer | string;
}
interface EntityChangeInput {
entityChange: EntityChange;
entity: EntityRow;
}
interface UpdateContext { interface UpdateContext {
alreadyErased: number; alreadyErased: number;
@@ -22,7 +12,7 @@ interface UpdateContext {
updated: Record<string, string[]> updated: Record<string, string[]>
} }
function updateEntities(entityChanges: EntityChangeInput[], instanceId: string) { function updateEntities(entityChanges: EntityChangeRecord[], instanceId: string) {
if (entityChanges.length === 0) { if (entityChanges.length === 0) {
return; return;
} }
@@ -51,8 +41,10 @@ function updateEntities(entityChanges: EntityChangeInput[], instanceId: string)
atLeastOnePullApplied = true; atLeastOnePullApplied = true;
} }
if (entity) {
updateEntity(entityChange, entity, instanceId, updateContext); updateEntity(entityChange, entity, instanceId, updateContext);
} }
}
logUpdateContext(updateContext); logUpdateContext(updateContext);
} }

2
src/services/utils.ts
View File

@@ -238,7 +238,7 @@ function getNoteTitle(filePath: string, replaceUnderscoresWithSpaces: boolean, n
} }
} }
function timeLimit<T>(promise: Promise<T>, limitMs: number, errorMessage: string): Promise<T> { function timeLimit<T>(promise: Promise<T>, limitMs: number, errorMessage?: string): Promise<T> {
if (!promise || !promise.then) { // it's not actually a promise if (!promise || !promise.then) { // it's not actually a promise
return promise; return promise;
} }