import * as AWS from 'aws-sdk' ; import * as _ from 'lodash' ; import { CrawlInputWithId } from '../crawler/types' ; import { CONTEXT_TABLE_READ_CAPACITY , CONTEXT_TABLE_WRITE_CAPACITY , PARALLEL_URLS_TO_SYNC } from '../config/constants' ; import { dynamodbPaginatedRequest } from './pagination' ; const ddb = new AWS . DynamoDB ( ) ; const ddbDoc = new AWS . DynamoDB . DocumentClient ( ) ; enum VisitStatus { VISITED = 'TRUE' , NOT_VISITED = 'FALSE' } return export const createContextTable = async ( target , contextTableNamePrefix ) : Promise < string > => { const TableName = `${ contextTableNamePrefix }-${ target . crawlName }-${ target . crawlId }` ; console . log ( 'Creating context table' , TableName ) ; await ddb . createTable ( { TableName , AttributeDefinitions : [ { AttributeName : 'visited' , AttributeType : 'S' , } , { AttributeName : 'path' , AttributeType : 'S' , } , ] , KeySchema : [ { AttributeName : 'visited' , KeyType : 'HASH' , } , { AttributeName : 'path' , KeyType : 'RANGE' , } , ] , ProvisionedThroughput : { ReadCapacityUnits : CONTEXT_TABLE_READ_CAPACITY , WriteCapacityUnits : CONTEXT_TABLE_WRITE_CAPACITY , } , } ) . promise ( ) ; console . log ( 'Waiting for table' , TableName , 'to finish creation' ) ; await ddb . waitFor ( 'tableExists' , { TableName } ) . promise ( ) ; return TableName ; } ; /**
 * Deletes the context table
 */ export const deleteContextTable = async ( contextTableName ) => { await ddb . deleteTable ( { TableName : contextTableName , } ) . promise ( ) ; } ; /**
 * Add a path to the url queue. Skips paths that have already been visited
 */ export const queuePath = async ( contextTableName , path ) => { try { const item = ( await ddbDoc . get ( { TableName : contextTableName , Key : { visited : VisitStatus . VISITED , path , } , } ) . promise ( ) ) . Item ; if ( ! item ) { await ddbDoc . put ( { TableName : contextTableName , Item : { path , visited : VisitStatus . NOT_VISITED , } , } ) . promise ( ) ; } } catch ( e ) { console . warn ( 'Unable to queue' , path , e ) ; } } ; /**
 * Add a list of paths to the url queue. Skips paths that have already been visited.
 */ export const queuePaths = async ( contextTableName , paths ) => { for ( const pathGroup of _ . chunk ( paths , 20 ) ) { await Promise . all ( pathGroup . map ( async ( newPath ) => queuePath ( contextTableName , newPath ) ) ) ; } } ; /**
 * Read a batch of urls to visit from the context table
 */ export const readBatchOfUrlsToVisit = async ( contextTableName ) : Promise < string [ ] > => { const toVisitEntries = await dynamodbPaginatedRequest ( ddbDoc . query . bind ( ddbDoc ) , { TableName : contextTableName , ConsistentRead : true , KeyConditionExpression : '#visited = :visited' , ExpressionAttributeValues : { ':visited' : VisitStatus . NOT_VISITED , } , ExpressionAttributeNames : { '#visited' : 'visited' , } , Limit : 50 , } , async ( ) => { } , PARALLEL_URLS_TO_SYNC ) ; return toVisitEntries . map ( ( entry ) => entry . path ) ; } ; /**
 * Marks a path as visited in the context table by adding a visited=TRUE entry and deleting the visited=FALSE entry
 */ export const markPathAsVisited = async ( contextTableName , path ) => { await ddbDoc . put ( { TableName : contextTableName , Item : { visited : VisitStatus . VISITED , path , visitedAt : new Date ( ) . toISOString ( ) , } , } ) . promise ( ) ; await ddbDoc . delete ( { TableName : contextTableName , Key : { visited : VisitStatus . NOT_VISITED , path , } , } ) . promise ( ) ; } ;