Improve drive management

4 files changed, 65 insertions, 1 deletions

diff --git a/src/queue/index.ts b/src/queue/index.ts
index a010004f15..70233d8318 100644
--- a/src/queue/index.ts
+++ b/src/queue/index.ts
@@ -8,6 +8,7 @@ import { program } from '../argv';
 import processDeliver from './processors/deliver';
 import processInbox from './processors/inbox';
 import processDb from './processors/db';
+import procesObjectStorage from './processors/object-storage';
 import { queueLogger } from './logger';
 import { DriveFile } from '../models/entities/drive-file';
 
@@ -34,9 +35,12 @@ function renderError(e: Error): any {
 export const deliverQueue = initializeQueue('deliver');
 export const inboxQueue = initializeQueue('inbox');
 export const dbQueue = initializeQueue('db');
+export const objectStorageQueue = initializeQueue('objectStorage');
 
 const deliverLogger = queueLogger.createSubLogger('deliver');
 const inboxLogger = queueLogger.createSubLogger('inbox');
+const dbLogger = queueLogger.createSubLogger('db');
+const objectStorageLogger = queueLogger.createSubLogger('objectStorage');
 
 deliverQueue
 	.on('waiting', (jobId) => deliverLogger.debug(`waiting id=${jobId}`))
@@ -54,6 +58,22 @@ inboxQueue
 	.on('error', (job: any, err: Error) => inboxLogger.error(`error ${err}`, { job, e: renderError(err) }))
 	.on('stalled', (job) => inboxLogger.warn(`stalled id=${job.id} activity=${job.data.activity ? job.data.activity.id : 'none'}`));
 
+dbQueue
+	.on('waiting', (jobId) => dbLogger.debug(`waiting id=${jobId}`))
+	.on('active', (job) => dbLogger.debug(`active id=${job.id}`))
+	.on('completed', (job, result) => dbLogger.debug(`completed(${result}) id=${job.id}`))
+	.on('failed', (job, err) => dbLogger.warn(`failed(${err}) id=${job.id}`, { job, e: renderError(err) }))
+	.on('error', (job: any, err: Error) => dbLogger.error(`error ${err}`, { job, e: renderError(err) }))
+	.on('stalled', (job) => dbLogger.warn(`stalled id=${job.id}`));
+
+objectStorageQueue
+	.on('waiting', (jobId) => objectStorageLogger.debug(`waiting id=${jobId}`))
+	.on('active', (job) => objectStorageLogger.debug(`active id=${job.id}`))
+	.on('completed', (job, result) => objectStorageLogger.debug(`completed(${result}) id=${job.id}`))
+	.on('failed', (job, err) => objectStorageLogger.warn(`failed(${err}) id=${job.id}`, { job, e: renderError(err) }))
+	.on('error', (job: any, err: Error) => objectStorageLogger.error(`error ${err}`, { job, e: renderError(err) }))
+	.on('stalled', (job) => objectStorageLogger.warn(`stalled id=${job.id}`));
+
 export function deliver(user: ILocalUser, content: any, to: any) {
 	if (content == null) return null;
 
@@ -165,11 +185,21 @@ export function createImportUserListsJob(user: ILocalUser, fileId: DriveFile['id
 	});
 }
 
+export function createDeleteObjectStorageFileJob(key: string) {
+	return objectStorageQueue.add('deleteFile', {
+		key: key
+	}, {
+		removeOnComplete: true,
+		removeOnFail: true
+	});
+}
+
 export default function() {
 	if (!program.onlyServer) {
 		deliverQueue.process(128, processDeliver);
 		inboxQueue.process(128, processInbox);
 		processDb(dbQueue);
+		procesObjectStorage(objectStorageQueue);
 	}
 }
 
diff --git a/src/queue/processors/db/delete-drive-files.ts b/src/queue/processors/db/delete-drive-files.ts
index 4e4eab86b7..5ecee9139a 100644
--- a/src/queue/processors/db/delete-drive-files.ts
+++ b/src/queue/processors/db/delete-drive-files.ts
@@ -1,7 +1,7 @@
 import * as Bull from 'bull';
 
 import { queueLogger } from '../../logger';
-import deleteFile from '../../../services/drive/delete-file';
+import { deleteFile } from '../../../services/drive/delete-file';
 import { Users, DriveFiles } from '../../../models';
 import { MoreThan } from 'typeorm';
 
diff --git a/src/queue/processors/object-storage/delete-file.ts b/src/queue/processors/object-storage/delete-file.ts
new file mode 100644
index 0000000000..8e6b5f959e
--- /dev/null
+++ b/src/queue/processors/object-storage/delete-file.ts
@@ -0,0 +1,22 @@
+import * as Bull from 'bull';
+import * as Minio from 'minio';
+import { fetchMeta } from '../../../misc/fetch-meta';
+
+export default async (job: Bull.Job) => {
+	const meta = await fetchMeta();
+
+	const minio = new Minio.Client({
+		endPoint: meta.objectStorageEndpoint!,
+		region: meta.objectStorageRegion ? meta.objectStorageRegion : undefined,
+		port: meta.objectStoragePort ? meta.objectStoragePort : undefined,
+		useSSL: meta.objectStorageUseSSL,
+		accessKey: meta.objectStorageAccessKey!,
+		secretKey: meta.objectStorageSecretKey!,
+	});
+
+	const key: string = job.data.key;
+
+	await minio.removeObject(meta.objectStorageBucket!, key);
+
+	return 'Success';
+};
diff --git a/src/queue/processors/object-storage/index.ts b/src/queue/processors/object-storage/index.ts
new file mode 100644
index 0000000000..f2fc08f1ab
--- /dev/null
+++ b/src/queue/processors/object-storage/index.ts
@@ -0,0 +1,12 @@
+import * as Bull from 'bull';
+import deleteFile from './delete-file';
+
+const jobs = {
+	deleteFile,
+} as any;
+
+export default function(q: Bull.Queue) {
+	for (const [k, v] of Object.entries(jobs)) {
+		q.process(k, v as any);
+	}
+}