Server: Improved share service reliability and optimised performance

packages/server/src/db.ts

@@ -15,6 +15,14 @@ require('pg').types.setTypeParser(20, function(val: any) {
 	return parseInt(val, 10);
 });
 
+// Also need this to get integers for count() queries.
+// https://knexjs.org/#Builder-count
+declare module 'knex/types/result' {
+    interface Registry {
+        Count: number;
+    }
+}
+
 const logger = Logger.create('db');
 
 // To prevent error "SQLITE_ERROR: too many SQL variables", SQL statements with

packages/server/src/models/BaseModel.ts

@@ -11,6 +11,8 @@ import Logger from '@joplin/lib/Logger';
 
 const logger = Logger.create('BaseModel');
 
+type SavePoint = string;
+
 export interface SaveOptions {
 	isNew?: boolean;
 	skipValidation?: boolean;
@@ -49,6 +51,7 @@ export default abstract class BaseModel<T> {
 	private modelFactory_: Function;
 	private static eventEmitter_: EventEmitter = null;
 	private config_: Config;
+	private savePoints_: SavePoint[] = [];
 
 	public constructor(db: DbConnection, modelFactory: Function, config: Config) {
 		this.db_ = db;
@@ -289,6 +292,25 @@ export default abstract class BaseModel<T> {
 		return this.db(this.tableName).select(options.fields || this.defaultFields).whereIn('id', ids);
 	}
 
+	public async setSavePoint(): Promise<SavePoint> {
+		const name = `sp_${uuidgen()}`;
+		await this.db.raw(`SAVEPOINT ${name}`);
+		this.savePoints_.push(name);
+		return name;
+	}
+
+	public async rollbackSavePoint(savePoint: SavePoint) {
+		const last = this.savePoints_.pop();
+		if (last !== savePoint) throw new Error('Rollback save point does not match');
+		await this.db.raw(`ROLLBACK TO SAVEPOINT ${savePoint}`);
+	}
+
+	public async releaseSavePoint(savePoint: SavePoint) {
+		const last = this.savePoints_.pop();
+		if (last !== savePoint) throw new Error('Rollback save point does not match');
+		await this.db.raw(`RELEASE SAVEPOINT ${savePoint}`);
+	}
+
 	public async exists(id: string): Promise<boolean> {
 		const o = await this.load(id, { fields: ['id'] });
 		return !!o;

packages/server/src/models/ItemModel.ts

@@ -106,13 +106,18 @@ export default class ItemModel extends BaseModel<Item> {
 		return path.replace(extractNameRegex, '$1');
 	}
 
-	public async byShareId(shareId: Uuid, options: LoadOptions = {}): Promise<Item[]> {
+	public byShareIdQuery(shareId: Uuid, options: LoadOptions = {}): Knex.QueryBuilder {
 		return this
 			.db('items')
 			.select(this.selectFields(options, null, 'items'))
 			.where('jop_share_id', '=', shareId);
 	}
 
+	public async byShareId(shareId: Uuid, options: LoadOptions = {}): Promise<Item[]> {
+		const query = this.byShareIdQuery(shareId, options);
+		return await query;
+	}
+
 	public async loadByJopIds(userId: Uuid | Uuid[], jopIds: string[], options: LoadOptions = {}): Promise<Item[]> {
 		if (!jopIds.length) return [];
 

packages/server/src/models/ShareModel.test.ts

@@ -1,7 +1,7 @@
-import { createUserAndSession, beforeAllDb, afterAllTests, beforeEachDb, models, checkThrowAsync, createItem, createItemTree } from '../utils/testing/testUtils';
+import { createUserAndSession, beforeAllDb, afterAllTests, beforeEachDb, models, checkThrowAsync, createItem, createItemTree, expectNotThrow, createNote } from '../utils/testing/testUtils';
 import { ErrorBadRequest, ErrorNotFound } from '../utils/errors';
 import { ShareType } from '../services/database/types';
-import { shareWithUserAndAccept } from '../utils/testing/shareApiUtils';
+import { inviteUserToShare, shareFolderWithUser, shareWithUserAndAccept } from '../utils/testing/shareApiUtils';
 
 describe('ShareModel', function() {
 
@@ -99,5 +99,80 @@ describe('ShareModel', function() {
 		expect(await models().item().load(noteItem.id)).toBeFalsy();
 	});
 
+	test('should count number of items in share', async function() {
+		const { user: user1, session: session1 } = await createUserAndSession(1);
+		const { session: session2 } = await createUserAndSession(2);
+
+		const { share } = await shareFolderWithUser(session1.id, session2.id, '000000000000000000000000000000F1', {
+			'000000000000000000000000000000F1': {
+				'00000000000000000000000000000001': null,
+			},
+		});
+
+		expect(await models().share().itemCountByShareId(share.id)).toBe(2);
+
+		await models().item().delete((await models().item().loadByJopId(user1.id, '00000000000000000000000000000001')).id);
+		await models().item().delete((await models().item().loadByJopId(user1.id, '000000000000000000000000000000F1')).id);
+
+		expect(await models().share().itemCountByShareId(share.id)).toBe(0);
+	});
+
+	test('should count number of items in share per recipient', async function() {
+		const { user: user1, session: session1 } = await createUserAndSession(1);
+		const { user: user2, session: session2 } = await createUserAndSession(2);
+		const { user: user3 } = await createUserAndSession(3);
+		await createUserAndSession(4); // To check that he's not included in the results since the items are not shared with him
+
+		const { share } = await shareFolderWithUser(session1.id, session2.id, '000000000000000000000000000000F1', {
+			'000000000000000000000000000000F1': {
+				'00000000000000000000000000000001': null,
+			},
+		});
+
+		await inviteUserToShare(share, session1.id, user3.email);
+
+		const rows = await models().share().itemCountByShareIdPerUser(share.id);
+
+		expect(rows.length).toBe(3);
+		expect(rows.find(r => r.user_id === user1.id).item_count).toBe(2);
+		expect(rows.find(r => r.user_id === user2.id).item_count).toBe(2);
+		expect(rows.find(r => r.user_id === user3.id).item_count).toBe(2);
+	});
+
+	test('should create user items for shared folder', async function() {
+		const { session: session1 } = await createUserAndSession(1);
+		const { session: session2 } = await createUserAndSession(2);
+		const { user: user3 } = await createUserAndSession(3);
+		await createUserAndSession(4); // To check that he's not included in the results since the items are not shared with him
+
+		const { share } = await shareFolderWithUser(session1.id, session2.id, '000000000000000000000000000000F1', {
+			'000000000000000000000000000000F1': {
+				'00000000000000000000000000000001': null,
+			},
+		});
+
+		// When running that function with a new user, it should get all the
+		// share items
+		expect((await models().userItem().byUserId(user3.id)).length).toBe(0);
+		await models().share().createSharedFolderUserItems(share.id, user3.id);
+		expect((await models().userItem().byUserId(user3.id)).length).toBe(2);
+
+		// Calling the function again should not throw - it should just ignore
+		// the items that have already been added.
+		await expectNotThrow(async () => models().share().createSharedFolderUserItems(share.id, user3.id));
+
+		// After adding a new note to the share, and calling the function, it
+		// should add the note to the other user collection.
+		expect(await models().share().itemCountByShareId(share.id)).toBe(2);
+
+		await createNote(session1.id, {
+			id: '00000000000000000000000000000003',
+			share_id: share.id,
+		});
+
+		expect(await models().share().itemCountByShareId(share.id)).toBe(3);
+		await models().share().createSharedFolderUserItems(share.id, user3.id);
+		expect(await models().share().itemCountByShareId(share.id)).toBe(3);
+	});
 
 });

packages/server/src/models/ShareModel.ts

@@ -8,6 +8,9 @@ import BaseModel, { AclAction, DeleteOptions, ValidateOptions } from './BaseMode
 import { userIdFromUserContentUrl } from '../utils/routeUtils';
 import { getCanShareFolder } from './utils/user';
 import { isUniqueConstraintError } from '../db';
+import Logger from '@joplin/lib/Logger';
+
+const logger = Logger.create('ShareModel');
 
 export default class ShareModel extends BaseModel<Share> {
 
@@ -215,6 +218,32 @@ export default class ShareModel extends BaseModel<Share> {
 			}
 		};
 
+		// This function add any missing item to a user's collection. Normally
+		// it shouldn't be necessary since items are added or removed based on
+		// the Change events, but it seems it can happen anyway, possibly due to
+		// a race condition somewhere. So this function corrects this by
+		// re-assigning any missing items.
+		//
+		// It should be relatively quick to call since it's restricted to shares
+		// that have recently changed, and the performed SQL queries are
+		// index-based.
+		const checkForMissingUserItems = async (shares: Share[]) => {
+			for (const share of shares) {
+				const realShareItemCount = await this.itemCountByShareId(share.id);
+				const shareItemCountPerUser = await this.itemCountByShareIdPerUser(share.id);
+
+				for (const row of shareItemCountPerUser) {
+					if (row.item_count < realShareItemCount) {
+						logger.warn(`checkForMissingUserItems: User is missing some items: Share ${share.id}: User ${row.user_id}`);
+						await this.createSharedFolderUserItems(share.id, row.user_id);
+					} else if (row.item_count > realShareItemCount) {
+						// Shouldn't be possible but log it just in case
+						logger.warn(`checkForMissingUserItems: User has too many items (??): Share ${share.id}: User ${row.user_id}`);
+					}
+				}
+			}
+		};
+
 		// This loop essentially applies the change made by one user to all the
 		// other users in the share.
 		//
@@ -260,6 +289,8 @@ export default class ShareModel extends BaseModel<Share> {
 						// too.
 					}
 
+					await checkForMissingUserItems(shares);
+
 					await this.models().keyValue().setValue('ShareService::latestProcessedChange', paginatedChanges.cursor);
 				}, 'ShareService::updateSharedItems3');
 			}
@@ -304,18 +335,13 @@ export default class ShareModel extends BaseModel<Share> {
 		}
 	}
 
-	// That should probably only be called when a user accepts the share
-	// invitation. At this point, we want to share all the items immediately.
-	// Afterwards, items that are added or removed are processed by the share
-	// service.
+	// The items that are added or removed from a share are processed by the
+	// share service, and added as user_utems to each user. This function
+	// however can be called after a user accept a share, or to correct share
+	// errors, but re-assigning all items to a user.
 	public async createSharedFolderUserItems(shareId: Uuid, userId: Uuid) {
-		const items = await this.models().item().byShareId(shareId, { fields: ['id'] });
-
-		await this.withTransaction(async () => {
-			for (const item of items) {
-				await this.models().userItem().add(userId, item.id);
-			}
-		}, 'ShareModel::createSharedFolderUserItems');
+		const query = this.models().item().byShareIdQuery(shareId, { fields: ['id', 'name'] });
+		await this.models().userItem().addMulti(userId, query);
 	}
 
 	public async shareFolder(owner: User, folderId: string): Promise<Share> {
@@ -368,4 +394,23 @@ export default class ShareModel extends BaseModel<Share> {
 		}, 'ShareModel::delete');
 	}
 
+	public async itemCountByShareId(shareId: Uuid): Promise<number> {
+		const r = await this
+			.db('items')
+			.count('id', { as: 'item_count' })
+			.where('jop_share_id', '=', shareId);
+		return r[0].item_count;
+	}
+
+	public async itemCountByShareIdPerUser(shareId: Uuid): Promise<{ item_count: number; user_id: Uuid }[]> {
+		return this.db('user_items')
+			.select(this.db.raw('user_id, count(user_id) as item_count'))
+			.whereIn('item_id',
+				this.db('items')
+					.select('id')
+					.where('jop_share_id', '=', shareId)
+			).groupBy('user_id') as any;
+	}
+
+
 }

packages/server/src/models/UserItemModel.ts

@@ -1,7 +1,8 @@
-import { ChangeType, ItemType, UserItem, Uuid } from '../services/database/types';
+import { ChangeType, Item, ItemType, UserItem, Uuid } from '../services/database/types';
 import BaseModel, { DeleteOptions, LoadOptions, SaveOptions } from './BaseModel';
 import { unique } from '../utils/array';
 import { ErrorNotFound } from '../utils/errors';
+import { Knex } from 'knex';
 
 interface DeleteByShare {
 	id: Uuid;
@@ -123,25 +124,38 @@ export default class UserItemModel extends BaseModel<UserItem> {
 		await this.deleteBy({ byShareId: shareId, byUserId: userId });
 	}
 
+	public async addMulti(userId: Uuid, itemsQuery: Knex.QueryBuilder | Item[], options: SaveOptions = {}): Promise<void> {
+		const items: Item[] = Array.isArray(itemsQuery) ? itemsQuery : await itemsQuery.whereNotIn('id', this.db('user_items').select('item_id').where('user_id', '=', userId));
+		if (!items.length) return;
+
+		await this.withTransaction(async () => {
+			for (const item of items) {
+				if (!('name' in item) || !('id' in item)) throw new Error('item.id and item.name must be set');
+
+				await super.save({
+					user_id: userId,
+					item_id: item.id,
+				}, options);
+
+				if (this.models().item().shouldRecordChange(item.name)) {
+					await this.models().change().save({
+						item_type: ItemType.UserItem,
+						item_id: item.id,
+						item_name: item.name,
+						type: ChangeType.Create,
+						previous_item: '',
+						user_id: userId,
+					});
+				}
+			}
+		}, 'UserItemModel::addMulti');
+	}
+
 	public async save(userItem: UserItem, options: SaveOptions = {}): Promise<UserItem> {
 		if (userItem.id) throw new Error('User items cannot be modified (only created or deleted)'); // Sanity check - shouldn't happen
-
 		const item = await this.models().item().load(userItem.item_id, { fields: ['id', 'name'] });
-
-		return this.withTransaction(async () => {
-			if (this.models().item().shouldRecordChange(item.name)) {
-				await this.models().change().save({
-					item_type: ItemType.UserItem,
-					item_id: userItem.item_id,
-					item_name: item.name,
-					type: ChangeType.Create,
-					previous_item: '',
-					user_id: userItem.user_id,
-				});
-			}
-
-			return super.save(userItem, options);
-		}, 'UserItemModel::save');
+		await this.addMulti(userItem.user_id, [item], options);
+		return this.byUserAndItemId(userItem.user_id, item.id);
 	}
 
 	public async delete(_id: string | string[], _options: DeleteOptions = {}): Promise<void> {

packages/server/src/utils/testing/shareApiUtils.ts

@@ -66,6 +66,21 @@ async function createItemTree3(sessionId: Uuid, userId: Uuid, parentFolderId: st
 	}
 }
 
+export async function inviteUserToShare(share: Share, sharerSessionId: string, recipientEmail: string, acceptShare: boolean = true) {
+	let shareUser = await postApi(sharerSessionId, `shares/${share.id}/users`, {
+		email: recipientEmail,
+	}) as ShareUser;
+
+	shareUser = await models().shareUser().load(shareUser.id);
+
+	if (acceptShare) {
+		const session = await models().session().createUserSession(shareUser.user_id);
+		await patchApi(session.id, `share_users/${shareUser.id}`, { status: ShareUserStatus.Accepted });
+	}
+
+	return shareUser;
+}
+
 export async function shareFolderWithUser(sharerSessionId: string, shareeSessionId: string, sharedFolderId: string, itemTree: any, acceptShare: boolean = true): Promise<ShareResult> {
 	itemTree = Array.isArray(itemTree) ? itemTree : convertTree(itemTree);
 
@@ -93,15 +108,7 @@ export async function shareFolderWithUser(sharerSessionId: string, shareeSession
 		}
 	}
 
-	let shareUser = await postApi(sharerSessionId, `shares/${share.id}/users`, {
-		email: sharee.email,
-	}) as ShareUser;
-
-	shareUser = await models().shareUser().load(shareUser.id);
-
-	if (acceptShare) {
-		await patchApi(shareeSessionId, `share_users/${shareUser.id}`, { status: ShareUserStatus.Accepted });
-	}
+	const shareUser = await inviteUserToShare(share, sharerSessionId, sharee.email, acceptShare);
 
 	await models().share().updateSharedItems3();
 
@@ -146,7 +153,6 @@ export async function shareWithUserAndAccept(sharerSessionId: string, shareeSess
 	await patchApi(shareeSessionId, `share_users/${shareUser.id}`, { status: ShareUserStatus.Accepted });
 
 	await models().share().updateSharedItems3();
-	// await models().share().updateSharedItems2(sharee.id);
 
 	return { share, item, shareUser };
 }