Improve drive management

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);
 	}
 }