self-hosted/joplin
1
0
Fork 0
mirror of https://github.com/laurent22/joplin.git synced 2026-06-18 20:16:34 +02:00
Code Issues Releases Activity

Server: Fixed change processing logic

Browse Source
This commit is contained in:
Laurent Cozic
2021-07-09 16:52:23 +01:00
parent bc97bb242a
commit 5a27d4dc31
5 changed files with 117 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/src/models/ChangeModel.test.ts
+35 -1
View File
@@ -1,4 +1,4 @@
import { createUserAndSession, beforeAllDb, afterAllTests, beforeEachDb, models, expectThrow, createFolder } from '../utils/testing/testUtils';
import { createUserAndSession, beforeAllDb, afterAllTests, beforeEachDb, models, expectThrow, createFolder, createItemTree3 } from '../utils/testing/testUtils';
import { ChangeType, Item, Uuid } from '../db';
import { msleep } from '../utils/time';
import { ChangePagination } from './ChangeModel';
@@ -130,4 +130,38 @@ describe('ChangeModel', function() {
await expectThrow(async () => changeModel.delta(user.id, { limit: 1, cursor: 'invalid' }), 'resyncRequired');
});
test('should tell that there are more changes even when current page is empty', async function() {
const { user: user1 } = await createUserAndSession(1);
const changeCount = 10;
const itemsToCreate: any[] = [];
for (let i = 0; i < changeCount / 2; i++) {
itemsToCreate.push({
id: (`${i}`).padStart(32, '0'),
children: [],
});
}
await createItemTree3(user1.id, '', '', itemsToCreate);
await models().item().deleteAll(user1.id);
expect((await models().change().all()).length).toBe(changeCount);
// Since all items have been deleted, the first change page is empty.
// However the "hasMore" property should be true to tell caller that
// they should fetch more changes.
const allFromIds1 = await models().change().allFromId('', changeCount / 2);
expect(allFromIds1.items.length).toBe(0);
expect(allFromIds1.has_more).toBe(true);
const allFromIds2 = await models().change().allFromId(allFromIds1.cursor, changeCount / 2);
expect(allFromIds2.items.length).toBe(5);
expect(allFromIds2.has_more).toBe(true);
const allFromIds3 = await models().change().allFromId(allFromIds2.cursor, changeCount / 2);
expect(allFromIds3.items.length).toBe(0);
expect(allFromIds3.has_more).toBe(false);
});
});
packages/server/src/models/ChangeModel.ts
+17 -7
View File
@@ -9,10 +9,14 @@ export interface DeltaChange extends Change {
jop_updated_time?: number;
}
export interface PaginatedChanges extends PaginatedResults {
export interface PaginatedDeltaChanges extends PaginatedResults {
items: DeltaChange[];
}
export interface PaginatedChanges extends PaginatedResults {
items: Change[];
}
export interface ChangePagination {
limit?: number;
cursor?: string;
@@ -55,15 +59,21 @@ export default class ChangeModel extends BaseModel<Change> {
return `${this.baseUrl}/changes`;
}
public async allFromId(id: string): Promise<Change[]> {
public async allFromId(id: string, limit: number = 1000): Promise<PaginatedChanges> {
const startChange: Change = id ? await this.load(id) : null;
const query = this.db(this.tableName).select(...this.defaultFields);
if (startChange) void query.where('counter', '>', startChange.counter);
void query.limit(1000);
let results = await query;
void query.limit(limit);
let results: Change[] = await query;
const hasMore = !!results.length;
const cursor = results.length ? results[results.length - 1].id : id;
results = await this.removeDeletedItems(results);
results = await this.compressChanges(results);
return results;
return {
items: results,
has_more: hasMore,
cursor,
};
}
private changesForUserQuery(userId: Uuid, count: boolean): Knex.QueryBuilder {
@@ -103,7 +113,7 @@ export default class ChangeModel extends BaseModel<Change> {
return query;
}
public async allByUser(userId: Uuid, pagination: Pagination = null): Promise<PaginatedChanges> {
public async allByUser(userId: Uuid, pagination: Pagination = null): Promise<PaginatedDeltaChanges> {
pagination = {
page: 1,
limit: 100,
@@ -132,7 +142,7 @@ export default class ChangeModel extends BaseModel<Change> {
};
}
public async delta(userId: Uuid, pagination: ChangePagination = null): Promise<PaginatedChanges> {
public async delta(userId: Uuid, pagination: ChangePagination = null): Promise<PaginatedDeltaChanges> {
pagination = {
...defaultDeltaPagination(),
...pagination,
packages/server/src/models/ItemModel.ts
+32 -23
View File
@@ -612,35 +612,44 @@ export default class ItemModel extends BaseModel<Item> {
while (true) {
const latestProcessedChange = await this.models().keyValue().value<string>('ItemModel::updateTotalSizes::latestProcessedChange');
const changes = await this.models().change().allFromId(latestProcessedChange || '');
if (!changes.length) break;
const paginatedChanges = await this.models().change().allFromId(latestProcessedChange || '');
const changes = paginatedChanges.items;
const itemIds: Uuid[] = unique(changes.map(c => c.item_id));
const userItems: UserItem[] = await this.db('user_items').select('user_id').whereIn('item_id', itemIds);
const userIds: Uuid[] = unique(userItems.map(u => u.user_id));
if (!changes.length) {
// `allFromId()` may return empty pages when all items have
// been deleted. In that case, we only save the cursor and
// continue.
await this.models().keyValue().setValue('ItemModel::updateTotalSizes::latestProcessedChange', paginatedChanges.cursor);
} else {
const itemIds: Uuid[] = unique(changes.map(c => c.item_id));
const userItems: UserItem[] = await this.db('user_items').select('user_id').whereIn('item_id', itemIds);
const userIds: Uuid[] = unique(userItems.map(u => u.user_id));
const totalSizes: TotalSizeRow[] = [];
for (const userId of userIds) {
if (doneUserIds[userId]) continue;
const totalSizes: TotalSizeRow[] = [];
for (const userId of userIds) {
if (doneUserIds[userId]) continue;
totalSizes.push({
userId,
totalSize: await this.calculateUserTotalSize(userId),
});
doneUserIds[userId] = true;
}
await this.withTransaction(async () => {
for (const row of totalSizes) {
await this.models().user().save({
id: row.userId,
total_item_size: row.totalSize,
totalSizes.push({
userId,
totalSize: await this.calculateUserTotalSize(userId),
});
doneUserIds[userId] = true;
}
await this.models().keyValue().setValue('ItemModel::updateTotalSizes::latestProcessedChange', changes[changes.length - 1].id);
}, 'ItemModel::updateTotalSizes');
await this.withTransaction(async () => {
for (const row of totalSizes) {
await this.models().user().save({
id: row.userId,
total_item_size: row.totalSize,
});
}
await this.models().keyValue().setValue('ItemModel::updateTotalSizes::latestProcessedChange', paginatedChanges.cursor);
}, 'ItemModel::updateTotalSizes');
}
if (!paginatedChanges.has_more) break;
}
} finally {
this.updatingTotalSizes_ = false;
packages/server/src/models/ShareModel.ts
+26 -20
View File
@@ -232,32 +232,38 @@ export default class ShareModel extends BaseModel<Share> {
while (true) {
const latestProcessedChange = await this.models().keyValue().value<string>('ShareService::latestProcessedChange');
const changes = await this.models().change().allFromId(latestProcessedChange || '');
if (!changes.length) break;
const paginatedChanges = await this.models().change().allFromId(latestProcessedChange || '');
const changes = paginatedChanges.items;
const items = await this.models().item().loadByIds(changes.map(c => c.item_id));
const shareIds = unique(items.filter(i => !!i.jop_share_id).map(i => i.jop_share_id));
const shares = await this.models().share().loadByIds(shareIds);
if (!changes.length) {
await this.models().keyValue().setValue('ShareService::latestProcessedChange', paginatedChanges.cursor);
} else {
const items = await this.models().item().loadByIds(changes.map(c => c.item_id));
const shareIds = unique(items.filter(i => !!i.jop_share_id).map(i => i.jop_share_id));
const shares = await this.models().share().loadByIds(shareIds);
await this.withTransaction(async () => {
for (const change of changes) {
const item = items.find(i => i.id === change.item_id);
await this.withTransaction(async () => {
for (const change of changes) {
const item = items.find(i => i.id === change.item_id);
if (change.type === ChangeType.Create) {
await handleCreated(change, item, shares.find(s => s.id === item.jop_share_id));
if (change.type === ChangeType.Create) {
await handleCreated(change, item, shares.find(s => s.id === item.jop_share_id));
}
if (change.type === ChangeType.Update) {
await handleUpdated(change, item, shares.find(s => s.id === item.jop_share_id));
}
// We don't need to handle ChangeType.Delete because when an
// item is deleted, all its associated userItems are deleted
// too.
}
if (change.type === ChangeType.Update) {
await handleUpdated(change, item, shares.find(s => s.id === item.jop_share_id));
}
await this.models().keyValue().setValue('ShareService::latestProcessedChange', paginatedChanges.cursor);
}, 'ShareService::updateSharedItems3');
}
// We don't need to handle ChangeType.Delete because when an
// item is deleted, all its associated userItems are deleted
// too.
}
await this.models().keyValue().setValue('ShareService::latestProcessedChange', changes[changes.length - 1].id);
});
if (!paginatedChanges.has_more) break;
}
}
packages/server/src/routes/api/shares.folder.test.ts
+7 -7
View File
@@ -1,7 +1,7 @@
import { ChangeType, Share, ShareType, ShareUser, ShareUserStatus } from '../../db';
import { beforeAllDb, afterAllTests, beforeEachDb, createUserAndSession, models, createNote, createFolder, updateItem, createItemTree, makeNoteSerializedBody, updateNote, expectHttpError, createResource } from '../../utils/testing/testUtils';
import { postApi, patchApi, getApi, deleteApi } from '../../utils/testing/apiUtils';
import { PaginatedChanges } from '../../models/ChangeModel';
import { PaginatedDeltaChanges } from '../../models/ChangeModel';
import { shareFolderWithUser } from '../../utils/testing/shareApiUtils';
import { msleep } from '../../utils/time';
import { ErrorForbidden } from '../../utils/errors';
@@ -477,11 +477,11 @@ describe('shares.folder', function() {
{
const names = ['000000000000000000000000000000F1.md', '00000000000000000000000000000001.md'].sort();
const page1 = await getApi<PaginatedChanges>(session1.id, 'items/root/delta');
const page1 = await getApi<PaginatedDeltaChanges>(session1.id, 'items/root/delta');
expect(page1.items.map(i => i.item_name).sort()).toEqual(names);
cursor1 = page1.cursor;
const page2 = await getApi<PaginatedChanges>(session2.id, 'items/root/delta');
const page2 = await getApi<PaginatedDeltaChanges>(session2.id, 'items/root/delta');
expect(page2.items.map(i => i.item_name).sort()).toEqual(names);
cursor2 = page2.cursor;
}
@@ -499,13 +499,13 @@ describe('shares.folder', function() {
}));
{
const page1 = await getApi<PaginatedChanges>(session1.id, 'items/root/delta', { query: { cursor: cursor1 } });
const page1 = await getApi<PaginatedDeltaChanges>(session1.id, 'items/root/delta', { query: { cursor: cursor1 } });
expect(page1.items.length).toBe(1);
expect(page1.items[0].item_name).toBe('00000000000000000000000000000001.md');
expect(page1.items[0].type).toBe(ChangeType.Update);
cursor1 = page1.cursor;
const page2 = await getApi<PaginatedChanges>(session2.id, 'items/root/delta', { query: { cursor: cursor2 } });
const page2 = await getApi<PaginatedDeltaChanges>(session2.id, 'items/root/delta', { query: { cursor: cursor2 } });
expect(page2.items.length).toBe(1);
expect(page2.items[0].item_name).toBe('00000000000000000000000000000001.md');
expect(page2.items[0].type).toBe(ChangeType.Update);
@@ -524,13 +524,13 @@ describe('shares.folder', function() {
}));
{
const page1 = await getApi<PaginatedChanges>(session1.id, 'items/root/delta', { query: { cursor: cursor1 } });
const page1 = await getApi<PaginatedDeltaChanges>(session1.id, 'items/root/delta', { query: { cursor: cursor1 } });
expect(page1.items.length).toBe(1);
expect(page1.items[0].item_name).toBe('00000000000000000000000000000001.md');
expect(page1.items[0].type).toBe(ChangeType.Update);
cursor1 = page1.cursor;
const page2 = await getApi<PaginatedChanges>(session2.id, 'items/root/delta', { query: { cursor: cursor2 } });
const page2 = await getApi<PaginatedDeltaChanges>(session2.id, 'items/root/delta', { query: { cursor: cursor2 } });
expect(page2.items.length).toBe(1);
expect(page2.items[0].item_name).toBe('00000000000000000000000000000001.md');
expect(page2.items[0].type).toBe(ChangeType.Update);