"use strict" ;
const Promise = require ( "bluebird" ) ;
const debug = require ( "debug" ) ( "srap:backend:postgresql:query:get-task-stream" ) ;
const buffer = require ( "@promistream/buffer" ) ;
const pipe = require ( "@promistream/pipe" ) ;
const simpleSource = require ( "@promistream/simple-source" ) ;
const fetchQuery = `
SELECT
srap _items . *
FROM
srap _queue
LEFT JOIN srap _items
ON srap _items . id = srap _queue . item _id
WHERE
srap _queue . task = : task
AND srap _queue . started = FALSE
LIMIT : resultLimit
` ;
const fillQuery = `
WITH
dependency _tasks AS (
SELECT * FROM
json _to _recordset ( : dependencyTaskDefinitions ) AS x ( task text , task _version text )
) ,
matching _items AS (
SELECT
srap _items . * ,
results . updated _at AS result _date ,
results . task _version ,
(
results . is _successful = TRUE
AND (
results . expires _at < NOW ( )
OR results . is _invalidated = TRUE
)
) AS is _candidate
FROM srap _items
INNER JOIN srap _tags
ON srap _tags . item _id = srap _items . id
AND srap _tags . name = ANY ( : tags )
LEFT JOIN srap _task _results AS results
ON results . item _id = srap _items . id
AND results . task = : task
) ,
candidates AS (
SELECT * FROM matching _items
WHERE result _date IS NULL
UNION ALL
SELECT * FROM matching _items
WHERE is _candidate = TRUE
OR NOT ( task _version = : taskVersion )
)
(
SELECT
: task AS task ,
id AS item _id
FROM
candidates
WHERE
NOT EXISTS (
SELECT
results . *
FROM dependency _tasks
LEFT JOIN srap _task _results AS results
ON dependency _tasks . task = results . task
AND dependency _tasks . task _version = results . task _version
AND results . item _id = candidates . id
WHERE
results . is _successful IS NULL
OR results . is _successful = FALSE
OR (
results . is _successful = TRUE
AND (
results . expires _at < NOW ( )
OR results . is _invalidated = TRUE
)
)
)
)
` ;
module . exports = function ( { metrics , backendSettings , knex } ) {
return function ( tx , { task } ) {
let refillParameters = {
tags : task . tags ,
task : task . name ,
taskVersion : task . version ,
dependencyTaskDefinitions : JSON . stringify ( task . dependencies . map ( ( dependency ) => {
// Case-mapping for SQL compatibility
return { task _version : dependency . version , task : dependency . name } ;
} ) )
} ;
let fetchParameters = {
task : task . name ,
resultLimit : backendSettings . taskBatchSize
} ;
function refillQueue ( ) {
let startTime = Date . now ( ) ;
return Promise . try ( ( ) => {
// NOTE: We are deliberately bypassing the transaction here! Also deliberately not using VALUES, since we're inserting from the results of another query instead
return knex . raw ( `
INSERT INTO srap _queue ( task , item _id )
( $ { fillQuery } )
ON CONFLICT ( task , item _id ) DO NOTHING ;
` , refillParameters);
} ) . then ( ( response ) => {
let timeElapsed = Date . now ( ) - startTime ;
metrics . taskRefillTime . labels ( { task : task . name } ) . set ( timeElapsed / 1000 ) ;
metrics . taskRefillResults . labels ( { task : task . name } ) . set ( response . rowCount ) ;
debug ( ` Queue for ' ${ task . name } ' was refilled with ${ response . rowCount } items in ${ timeElapsed } ms ` ) ;
return response . rowCount ;
} ) ;
}
return pipe ( [
simpleSource ( ( ) => {
let startTime = Date . now ( ) ;
return Promise . try ( ( ) => {
return tx . raw ( fetchQuery , fetchParameters ) ;
} ) . then ( ( result ) => {
let timeElapsed = Date . now ( ) - startTime ;
metrics . taskFetchTime . labels ( { task : task . name } ) . set ( timeElapsed / 1000 ) ;
metrics . taskFetchResults . labels ( { task : task . name } ) . set ( result . rowCount ) ;
debug ( ` Task retrieval query for ' ${ task . name } ' took ${ timeElapsed } ms and produced ${ result . rowCount } results ` ) ;
if ( result . rowCount > 0 ) {
return result . rows ;
} else {
return Promise . try ( ( ) => {
return refillQueue ( ) ;
} ) . then ( ( newItems ) => {
if ( newItems === 0 ) {
// TODO: Consider using LISTEN/NOTIFY instead? Worth the added complexity?
return Promise . resolve ( [ ] ) . delay ( backendSettings . taskBatchDelay ) ;
} else {
// Have another go right away
return [ ] ;
}
} ) ;
}
} ) ;
} ) ,
buffer ( )
] ) ;
} ;
} ;