log

packages/server/src/app.ts

@@ -3,7 +3,7 @@ require('source-map-support').install();
 
 import * as Koa from 'koa';
 import * as fs from 'fs-extra';
-import { argv } from 'yargs';
+import { argv as yargsArgv } from 'yargs';
 import Logger, { LoggerWrapper, TargetType } from '@joplin/lib/Logger';
 import config, { initConfig, runningInDocker, EnvVariables } from './config';
 import { createDb, dropDb } from './tools/dbTools';
@@ -19,6 +19,24 @@ import startServices from './utils/startServices';
 import { credentialFile } from './utils/testing/testUtils';
 import apiVersionHandler from './middleware/apiVersionHandler';
 import clickJackingHandler from './middleware/clickJackingHandler';
+import deleteOldChanges from './commands/deleteOldChanges';
+import newModelFactory from './models/factory';
+
+interface Argv {
+	env?: Env;
+	migrateLatest?: boolean;
+	migrateUp?: boolean;
+	migrateDown?: boolean;
+	migrateList?: boolean;
+	dropDb?: boolean;
+	pidfile?: string;
+	dropTables?: boolean;
+	createDb?: boolean;
+	envFile?: string;
+	deleteOldChanges?: boolean;
+}
+
+const argv: Argv = yargsArgv as any;
 
 const nodeSqlite = require('sqlite3');
 const cors = require('@koa/cors');
@@ -225,6 +243,13 @@ async function main() {
 		await disconnectDb(db);
 	} else if (argv.createDb) {
 		await createDb(config().database);
+	} else if (argv.deleteOldChanges) {
+		// Eventually all commands should be started in a more generic way. All
+		// should go under /commands, and they will receive a context object
+		// with an intialized models property.
+		const connectionCheck = await waitForConnection(config().database);
+		const models = newModelFactory(connectionCheck.connection, config());
+		await deleteOldChanges({ models });
 	} else {
 		runCommandAndExitApp = false;
 

packages/server/src/commands/deleteOldChanges.ts

@@ -0,0 +1,5 @@
+import { CommandContext } from '../utils/types';
+
+export default async function(ctx: CommandContext) {
+	await ctx.models.change().deleteOldChanges();
+}

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

@@ -1,4 +1,4 @@
-import { createUserAndSession, beforeAllDb, afterAllTests, beforeEachDb, models, expectThrow, createFolder, createItemTree3, expectNotThrow } from '../utils/testing/testUtils';
+import { createUserAndSession, beforeAllDb, afterAllTests, beforeEachDb, models, expectThrow, createFolder, createItemTree3, expectNotThrow, createNote, updateNote } from '../utils/testing/testUtils';
 import { ChangeType } from '../services/database/types';
 import { msleep } from '../utils/time';
 import { ChangePagination } from './ChangeModel';
@@ -178,4 +178,84 @@ describe('ChangeModel', function() {
 		expect(changeCount).toBe(SqliteMaxVariableNum);
 	});
 
+	test('should delete old changes', async function() {
+		// Create the following events:
+		//
+		// T1   2020-01-01    U1 Create
+		// T2   2020-01-10    U1 Update    U2 Create
+		// T3   2020-01-20    U1 Update
+		// T4   2020-01-30    U1 Update
+		// T5   2020-02-10                 U2 Update
+		// T6   2020-02-20                 U2 Update
+		//
+		// Use this to add days to a date:
+		//
+		// https://www.timeanddate.com/date/dateadd.html
+
+		const { session: session1 } = await createUserAndSession(1);
+		const { session: session2 } = await createUserAndSession(2);
+
+		jest.useFakeTimers('modern');
+
+		jest.setSystemTime(new Date('2020-01-01').getTime());
+		const note1 = await createNote(session1.id, {});
+
+		jest.setSystemTime(new Date('2020-01-10').getTime());
+		const note2 = await createNote(session2.id, {});
+		await updateNote(session1.id, { id: note1.jop_id });
+
+		jest.setSystemTime(new Date('2020-01-20').getTime());
+		await updateNote(session1.id, { id: note1.jop_id });
+
+		const t4 = new Date('2020-01-30').getTime();
+		jest.setSystemTime(t4);
+		await updateNote(session1.id, { id: note1.jop_id });
+
+		const t5 = new Date('2020-02-10').getTime();
+		jest.setSystemTime(t5);
+		await updateNote(session2.id, { id: note2.jop_id });
+
+		const t6 = new Date('2020-02-20').getTime();
+		jest.setSystemTime(t6);
+		await updateNote(session2.id, { id: note2.jop_id });
+
+		expect(await models().change().count()).toBe(7);
+
+		// Shouldn't do anything initially because it only deletes old changes.
+		await models().change().deleteOldChanges();
+		expect(await models().change().count()).toBe(7);
+
+		// 90 days later, it should delete all U1 updates events except for the
+		// last one
+		jest.setSystemTime(new Date('2020-05-01').getTime());
+		await models().change().deleteOldChanges();
+		expect(await models().change().count()).toBe(5);
+		{
+			const updateChange = (await models().change().all()).find(c => c.item_id === note1.id && c.type === ChangeType.Update);
+			expect(updateChange.created_time >= t4 && updateChange.created_time < t5).toBe(true);
+		}
+		// None of the note 2 changes should have been deleted because they've
+		// been made later
+		expect((await models().change().all()).filter(c => c.item_id === note2.id).length).toBe(3);
+
+		// Between T5 and T6, 90 days later - nothing should happen because
+		// there's only one note 2 change that is older than 90 days at this
+		// point.
+		jest.setSystemTime(new Date('2020-05-15').getTime());
+		await models().change().deleteOldChanges();
+		expect(await models().change().count()).toBe(5);
+
+		// After T6, more than 90 days later - now the change at T5 should be
+		// deleted, keeping only the change at T6.
+		jest.setSystemTime(new Date('2020-05-25').getTime());
+		await models().change().deleteOldChanges();
+		expect(await models().change().count()).toBe(4);
+		{
+			const updateChange = (await models().change().all()).find(c => c.item_id === note2.id && c.type === ChangeType.Update);
+			expect(updateChange.created_time >= t6).toBe(true);
+		}
+
+		jest.useRealTimers();
+	});
+
 });

packages/server/src/models/ChangeModel.ts

@@ -1,11 +1,17 @@
 import { Knex } from 'knex';
+import Logger from '@joplin/lib/Logger';
 import { SqliteMaxVariableNum } from '../db';
 import { Change, ChangeType, Item, Uuid } from '../services/database/types';
 import { md5 } from '../utils/crypto';
 import { ErrorResyncRequired } from '../utils/errors';
+import { Day, formatDateTime } from '../utils/time';
 import BaseModel, { SaveOptions } from './BaseModel';
 import { PaginatedResults, Pagination, PaginationOrderDir } from './utils/pagination';
 
+const logger = Logger.create('ChangeModel');
+
+export const changeTtl = 90 * Day;
+
 export interface DeltaChange extends Change {
 	jop_updated_time?: number;
 }
@@ -303,6 +309,71 @@ export default class ChangeModel extends BaseModel<Change> {
 		return output;
 	}
 
+	public async deleteOldChanges() {
+		const cutOffDate = Date.now() - changeTtl;
+		const limit = 1000;
+		const doneItemIds: Uuid[] = [];
+
+		interface ChangeReportItem {
+			total: number;
+			max_created_time: number;
+			item_id: Uuid;
+		}
+
+		let error: Error = null;
+		let totalDeletedCount = 0;
+
+		logger.info(`deleteOldChanges: Processing changes older than: ${formatDateTime(cutOffDate)} (${cutOffDate})`);
+
+		while (true) {
+			const changeReport: ChangeReportItem[] = await this
+				.db(this.tableName)
+
+				.select(['item_id'])
+				.countDistinct('id', { as: 'total' })
+				.max('created_time', { as: 'max_created_time' })
+
+				.where('type', '=', ChangeType.Update)
+				.where('created_time', '<', cutOffDate)
+
+				.groupBy('item_id')
+				.havingRaw('count(id) > 1')
+				.orderBy('total', 'desc')
+				.limit(limit);
+
+			if (!changeReport.length) break;
+
+			await this.withTransaction(async () => {
+				for (const row of changeReport) {
+					if (doneItemIds.includes(row.item_id)) {
+						// We don't throw from within the transaction because
+						// that would rollback all other operations even though
+						// they are valid. So we save the error and exit.
+						error = new Error(`Trying to process an item that has already been done. Aborting. Row: ${JSON.stringify(row)}`);
+						return;
+					}
+
+					const deletedCount = await this
+						.db(this.tableName)
+						.where('type', '=', ChangeType.Update)
+						.where('created_time', '<', cutOffDate)
+						.where('created_time', '!=', row.max_created_time)
+						.where('item_id', '=', row.item_id)
+						.delete();
+
+					totalDeletedCount += deletedCount;
+					doneItemIds.push(row.item_id);
+				}
+			}, 'ChangeModel::deleteOldChanges');
+
+			logger.info(`deleteOldChanges: Processed: ${doneItemIds.length} items. Deleted: ${totalDeletedCount}`);
+
+			if (error) throw error;
+		}
+
+		logger.info(`deleteOldChanges: Finished processing. Done ${doneItemIds.length} items. Deleted: ${totalDeletedCount}`);
+	}
+
 	public async save(change: Change, options: SaveOptions = {}): Promise<Change> {
 		const savedChange = await super.save(change, options);
 		ChangeModel.eventEmitter.emit('saved');

packages/server/src/utils/types.ts

@@ -131,3 +131,7 @@ export enum RouteType {
 }
 
 export type KoaNext = ()=> Promise<void>;
+
+export interface CommandContext {
+	models: Models;
+}