import { invalidateFQC } from '@matters/apollo-response-cache' ; import Queue from 'bull' ; import { APPRECIATION_TYPES , ARTICLE_STATE , DB_NOTICE_TYPE , NODE_TYPES , QUEUE_JOB , QUEUE_NAME , QUEUE_PRIORITY , } from 'common/enums' ; import { environment } from 'common/environment' ; import { ActionLimitExceededError , ArticleNotFoundError , ForbiddenError , UserNotFoundError , } from 'common/errors' ; import logger from 'common/logger' ; import { BaseQueue } from './baseQueue' ; import { likeCoinQueue } from './likecoin' ; interface AppreciationParams { amount : number ; articleId : string ; senderId : string ; senderIP ? : string ; userAgent : string ; } class AppreciationQueue extends BaseQueue { constructor ( ) { super ( QUEUE_NAME . appreciation , { limiter : { max : 1 , duration : 500 } } ) ; this . addConsumers ( ) ; } /**
     * Producer for appreciation.
     *
     */ appreciate = ( { amount , articleId , senderId , senderIP , userAgent , } ) => { return this . q . add ( QUEUE_JOB . appreciation , { amount , articleId , senderId , senderIP , userAgent } , { priority : QUEUE_PRIORITY . NORMAL , removeOnComplete : true , } ) ; } ; see https private addConsumers = ( ) => { this . q . process ( QUEUE_JOB . appreciation , 1 , this . handleAppreciation ) ; } ; /**
     * Appreciation handler.
     *
     */ private handleAppreciation : Queue . ProcessCallbackFunction < unknown > = async ( job , done ) => { try { const { amount , articleId , senderId , senderIP , userAgent } = job . data as AppreciationParams ; if ( ! articleId || ! senderId ) { throw new Error ( 'appreciation job has no required data' ) ; } const article = await this . atomService . findFirst ( { table : 'article' , where : { id : articleId , state : ARTICLE_STATE . active } , } ) ; if ( ! article ) { throw new ArticleNotFoundError ( 'article does not exist' ) ; } if ( article . authorId === senderId ) { throw new ForbiddenError ( 'cannot appreciate your own article' ) ; } const appreciateLeft = await this . articleService . appreciateLeftByUser ( { articleId , userId : senderId , } ) ; if ( appreciateLeft <= 0 ) { throw new ActionLimitExceededError ( 'too many appreciations' ) ; } const validAmount = Math . min ( amount , appreciateLeft ) ; const [ author , sender ] = await Promise . all ( [ this . userService . baseFindById ( article . authorId ) , this . userService . baseFindById ( senderId ) , ] ) ; if ( ! author || ! sender ) { throw new UserNotFoundError ( 'user not found' ) ; } if ( ! author . likerId || ! sender . likerId ) { throw new ForbiddenError ( 'user has no liker id' ) ; } await this . articleService . appreciate ( { articleId : article . id , senderId , recipientId : article . authorId , amount : validAmount , type : APPRECIATION_TYPES . like , } ) ; likeCoinQueue . like ( { likerId : sender . likerId , likerIp : senderIP , userAgent , authorLikerId : author . likerId , url : `${ environment . siteDomain }/@${ author . userName }/${ article . slug }-${ article . mediaHash }` , amount : validAmount , } ) ; this . notificationService . trigger ( { event : DB_NOTICE_TYPE . article_new_appreciation , actorId : sender . id , recipientId : author . id , entities : [ { type : 'target' , entityTable : 'article' , entity : article , } , ] , } ) ; if ( this . cacheService ) { invalidateFQC ( { node : { type : NODE_TYPES . Article , id : article . id } , redis : this . cacheService . redis , } ) ; invalidateFQC ( { node : { type : NODE_TYPES . User , id : article . authorId } , redis : this . cacheService . redis , } ) ; } job . progress ( 100 ) ; done ( null , job . data ) ; } catch ( error ) { logger . error ( error ) ; done ( error ) ; } } ; } export const appreciationQueue = new AppreciationQueue ( ) ;