import Mysql from './model/db/mysql' ; import Bucket from './bucket' ; import { S3 } from 'aws-sdk' ; import * as Bull from 'bull' ; import { Job , Queue , QueueOptions } from 'bull' ; import { getConfig } from '../config-helper' ; import { TakeoutRequest , TakeoutState } from 'common' ; const kTakeoutConcurrency = 10 ; const kTakeoutRateLimiter = { max : 10 , duration : 3600 } ; const kChunkMaxFiles = 2000 ; const kExpirationDays = 30 ; const kExpirationMarginHours = 3 ; const kStuckDurationHours = 24 ; export type TaskQueues = { dataTakeout : Queue < TakeoutTask > ; dataTakeoutCleanup : Queue < { } > ; } ; export function createTaskQueues ( takeout ) { const redisUrlParts = getConfig ( ) . REDIS_URL ?. split ( '//' ) ; const redisDomain = redisUrlParts . length > 1 ? redisUrlParts [ 1 ] : redisUrlParts [ 0 ] ; let redisOpts = { host : redisDomain } ; if ( getConfig ( ) . REDIS_URL . includes ( 'rediss://' ) ) { redisOpts = { ... redisOpts , tls : redisOpts } ; } const createQueue = < T > ( name , params ? ) => { const bull = new Bull < T > ( name , { redis : redisOpts , prefix : `bull-${ name }-` , ... params , } ) ; if ( getConfig ( ) . DEBUG ) { console . log ( bull ) ; } return bull ; } ; const dataTakeout = createQueue < TakeoutTask > ( 'data-takeout' , { limiter : kTakeoutRateLimiter , } ) ; dataTakeout . on ( 'error' , console . error ) ; dataTakeout . on ( 'failed' , console . error ) ; dataTakeout . process ( kTakeoutConcurrency , async ( job ) => await takeout . takeoutWorker ( job ) ) ; const dataTakeoutCleanup = createQueue < { } > ( 'data-takeout-cleanup' ) ; dataTakeoutCleanup . process ( 1 , async ( job ) => await takeout . takeoutCleanupWorker ( job ) ) ; dataTakeoutCleanup . add ( { } , { repeat : { cron : '0 15 * * * *' } } ) ; return { dataTakeout , dataTakeoutCleanup } ; } type TakeoutTask = { takeout_id : number ; } ; export type ClientClip = { original_sentence_id : string ; sentence : string ; path : string ; locale : string ; } ; export default class Takeout { private readonly db : Mysql ; private readonly s3 : S3 ; private readonly bucket : Bucket ; private queue : Bull . Queue < TakeoutTask > ; constructor ( db , s3 , bucket ) { this . db = db ; this . s3 = s3 ; this . bucket = bucket ; } setQueue ( queue ) { this . queue = queue ; } async startTakeout ( client_id ) { const takeout_id = await this . insertNewTakeout ( client_id ) ; await this . queue . add ( { takeout_id } , { removeOnComplete : true } ) ; return takeout_id ; } async generateDownloadLinks ( client_id , takeout_id ) { const takeout = await this . getTakeout ( takeout_id ) ; if ( takeout === null ) throw 'no such takeout' ; if ( takeout . client_id !== client_id ) throw 'no such takeout' ; if ( takeout . state !== TakeoutState . AVAILABLE ) throw 'takeout is unavailable' ; const keys = [ ... Array ( takeout . archive_count ) . keys ( ) ] . map ( offset => this . bucket . takeoutKey ( takeout , offset ) ) ; const links = [ keys . map ( key => this . bucket . getPublicUrl ( key ) ) , this . bucket . getPublicUrl ( this . bucket . metadataKey ( takeout ) ) , ] ; return links ; } async getClientClips ( client_id ) { const [ clips ] = await this . db . query ( `
        SELECT original_sentence_id, sentence, path, name AS locale FROM clips LEFT JOIN locales ON clips.locale_id = locales.id WHERE client_id = ?
      ` , [ client_id ] ) ; return clips ; } async takeoutWorker ( job ) { console . log ( 'performing takeout' , job . data . takeout_id ) ; const takeout_id = job . data . takeout_id ; const takeout = await this . getTakeout ( takeout_id ) ; if ( takeout === null ) throw 'unknown takeout' ; if ( takeout . state !== TakeoutState . PENDING ) { console . log ( `takeout ${ job . data . takeout_id } already in progress, do nothing` ) ; return ; } await this . markTakeoutInProgress ( takeout_id ) ; const clips = await this . getClientClips ( takeout . client_id ) ; const chunkedClips = Takeout . splitIntoChunks ( clips . map ( clip => clip . path ) ) ; const fileSizes = [ ] ; chunkedClips . forEach ( async ( chunk , index ) => { fileSizes . push ( await this . bucket . zipTakeoutFilesToS3 ( takeout , index , chunk ) ) ; await job . progress ( Math . ceil ( ( 100 * index ) / chunkedClips . length ) ) ; } ) ; fileSizes . push ( await this . bucket . uploadClipMetadata ( takeout , clips ) ) ; const totalSize = fileSizes . reduce ( ( acc , curr ) => ( acc += curr . ContentLength ) , 0 ) ; await this . finalizeTakeout ( takeout_id , clips . length , totalSize , chunkedClips . length ) ; } async takeoutCleanupWorker ( job ) { console . log ( 'Running takeout cleanup worker' ) ; const expired_deleted = await this . deleteCloseToExpirationTakeouts ( ) ; if ( expired_deleted ) console . log ( `deleted ${ expired_deleted } expired takeouts` ) ; const stuck_deleted = await this . deleteStuckTakeouts ( ) ; if ( stuck_deleted ) console . log ( `deleted ${ stuck_deleted } stuck takeouts; you might want to investigate` ) ; } async getClientTakeouts ( client_id ) { const [ takeouts ] = ( await this . db . query ( `
      SELECT * FROM user_client_takeouts
      WHERE client_id = ? AND (expiration_date IS NULL OR NOW() < expiration_date) 
      ORDER BY requested_date DESC
    ` , [ client_id ] ) ) as [ TakeoutRequest [ ] ] ; return takeouts ; } async getTakeout ( takeout_id ) { const [ takeouts ] = ( await this . db . query ( `
      SELECT * FROM user_client_takeouts
      WHERE id = ?
      LIMIT 1
    ` , [ takeout_id ] ) ) as [ TakeoutRequest [ ] ] ; return takeouts [ 0 ] || null ; } private async deleteTakeoutFiles ( takeouts ) { takeouts . forEach ( async ( takeout ) => { for ( let i = 0 ; i < takeout . archive_count ; i ++ ) { await this . bucket . deletePath ( this . bucket . takeoutKey ( takeout , i ) ) ; } await this . bucket . deletePath ( this . bucket . metadataKey ( takeout ) ) ; } ) ; console . log ( `deleted s3 files for the following takeout ids: ${ takeouts . map ( takeout => takeout . id ) . join ( ',' ) }` ) ; } private async deleteCloseToExpirationTakeouts ( ) { const [ expiredTakeouts ] = ( await this . db . query ( `
        SELECT * FROM user_client_takeouts
        WHERE expiration_date IS NOT NULL
          AND state != ?
          AND DATE_SUB(expiration_date, INTERVAL ? HOUR) < NOW()
      ` , [ TakeoutState . EXPIRED , kExpirationMarginHours ] ) ) as [ TakeoutRequest [ ] ] ; if ( expiredTakeouts . length ) await this . deleteTakeoutFiles ( expiredTakeouts ) ; return ( await this . db . query ( `
          UPDATE user_client_takeouts
          SET state = ?
          WHERE id IN (?)
        ` , [ TakeoutState . EXPIRED , expiredTakeouts . map ( takeout => takeout . id ) ] ) ) [ 0 ] . affectedRows ; } private async deleteStuckTakeouts ( ) { const [ stuckTakeouts ] = ( await this . db . query ( `
        SELECT * FROM user_client_takeouts
        WHERE state IN (?)
          AND DATE_ADD(requested_date, INTERVAL ? HOUR) < NOW()
      ` , [ [ TakeoutState . PENDING , TakeoutState . IN_PROGRESS ] , kStuckDurationHours ] ) ) as [ TakeoutRequest [ ] ] ; if ( stuckTakeouts . length ) await this . deleteTakeoutFiles ( stuckTakeouts ) ; return ( await this . db . query ( `
          UPDATE user_client_takeouts
          SET state = ?
          WHERE id IN (?)
        ` , [ TakeoutState . INCOMPLETE , stuckTakeouts . map ( takeout => takeout . id ) ] ) ) [ 0 ] . affectedRows ; } private async insertNewTakeout ( client_id ) { if ( ! ( await this . clientCanRequestTakeout ( client_id ) ) ) throw new Error ( 'pending takeout' ) ; const [ q ] = await this . db . query ( `
      INSERT INTO user_client_takeouts (client_id, state, requested_date)
      VALUES (?, ?, NOW())
    ` , [ client_id , TakeoutState . PENDING ] ) ; return q . insertId ; } /**
     * Returns whether a client can start a new takeout, which is only true when
     * there is no in-progress takeout already.
     */ private async clientCanRequestTakeout ( client_id ) { const [ pending ] = await this . db . query ( `
      SELECT 1 FROM user_client_takeouts
      WHERE client_id = ? AND state IN (?)
      LIMIT 1
    ` , [ client_id , [ TakeoutState . PENDING , TakeoutState . IN_PROGRESS ] ] ) ; return pending . length === 0 ; } private async finalizeTakeout ( takeout_id , clip_count , clip_total_size , archive_count ) { console . log ( 'finalizing takeout' , takeout_id ) ; return await this . db . query ( `
      UPDATE user_client_takeouts 
      SET 
        state = ?, 
        expiration_date = DATE_ADD(NOW(), INTERVAL ? DAY),
        clip_count = ?,
        clip_total_size = ?,
        archive_count = ?
      WHERE id = ?
    ` , [ TakeoutState . AVAILABLE , kExpirationDays , clip_count , clip_total_size , archive_count , takeout_id , ] ) ; } private async markTakeoutInProgress ( takeout_id ) { return await this . db . query ( `
      UPDATE user_client_takeouts
      SET state = ?
      WHERE id = ?
    ` , [ TakeoutState . IN_PROGRESS , takeout_id ] ) ; } private static splitIntoChunks ( paths ) { const todo = [ ... paths ] ; const chunks = [ ] ; do { const currentChunk = [ ] ; while ( todo . length && currentChunk . length < kChunkMaxFiles ) { const clip = todo . pop ( ) ; currentChunk . push ( clip ) ; } chunks . push ( currentChunk ) ; } while ( todo . length ) ; return chunks ; } }