import { QueryDocument , QueryModel } from "../models/QueryModel" ; import { MetricValueParams , SourceIntegrationInterface , ExperimentMetricQueryParams , MetricValueQueryResponse , MetricValueResult , PastExperimentResponse , PastExperimentResult , ExperimentResults , ExperimentQueryResponses , } from "../types/Integration" ; import uniqid from "uniqid" ; import { Queries , QueryInterface , QueryPointer , QueryStatus , } from "../../types/query" ; import { ExperimentInterface , ExperimentPhase } from "../../types/experiment" ; import { MetricInterface , MetricStats } from "../../types/metric" ; import { DimensionInterface } from "../../types/dimension" ; import { getValidDate } from "../util/dates" ; import { QUERY_CACHE_TTL_MINS } from "../util/secrets" ; export type QueryMap = Map < string , QueryInterface > ; export type InterfaceWithQueries = { runStarted : Date | null ; queries : Queries ; organization : string ; } ; async function getExistingQuery ( integration , query ) { const earliestDate = new Date ( ) ; earliestDate . setMinutes ( earliestDate . getMinutes ( ) - QUERY_CACHE_TTL_MINS ) ; const lastHeartbeat = new Date ( ) ; lastHeartbeat . setMinutes ( lastHeartbeat . getMinutes ( ) - 2 ) ; const existing = await QueryModel . find ( { organization : integration . organization , datasource : integration . datasource , query , createdAt : { $gt : earliestDate , } , status : { $in : [ "running" , "succeeded" ] , } , } ) . sort ( { createdAt : - 1 } ) . limit ( 5 ) ; for ( let i = 0 ; i < existing . length ; i ++ ) { if ( existing [ i ] . status === "succeeded" ) { return existing [ i ] ; } if ( existing [ i ] . heartbeat >= lastHeartbeat ) { return existing [ i ] ; } } return null ; } async function createNewQuery ( integration , query , result = null , error = null ) { const data = { createdAt : new Date ( ) , datasource : integration . datasource , finishedAt : result || error ? new Date ( ) : undefined , heartbeat : new Date ( ) , id : uniqid ( "qry_" ) , language : integration . getSourceProperties ( ) . queryLanguage , organization : integration . organization , query , startedAt : new Date ( ) , status : result ? "succeeded" : error ? "failed" : "running" , result : result || undefined , error : error || undefined , } ; return await QueryModel . create ( data ) ; } async function getQueryDoc < T , P > ( integration , query , run , process , useExisting = true ) { if ( useExisting ) { const existing = await getExistingQuery ( integration , query ) ; if ( existing ) return existing ; } const doc = await createNewQuery ( integration , query ) ; const timer = setInterval ( ( ) => { doc . set ( "heartbeat" , new Date ( ) ) ; doc . save ( ) ; } , 30000 ) ; run ( query ) . then ( ( rows ) => { clearInterval ( timer ) ; doc . set ( "finishedAt" , new Date ( ) ) ; doc . set ( "status" , "succeeded" ) ; doc . set ( "rawResult" , rows ) ; doc . set ( "result" , process ( rows ) ) ; doc . save ( ) ; } ) . catch ( ( e ) => { clearInterval ( timer ) ; doc . set ( "finishedAt" , new Date ( ) ) ; doc . set ( "status" , "failed" ) ; doc . set ( "error" , e . message ) ; doc . save ( ) ; } ) ; return doc ; } export async function getPastExperiments ( integration , from , minLength ? ) { return getQueryDoc ( integration , integration . getPastExperimentQuery ( { from , minLength , } ) , ( query ) => integration . runPastExperimentQuery ( query ) , processPastExperimentQueryResponse ) ; } export async function getMetricValue ( integration , params ) { return getQueryDoc ( integration , integration . getMetricValueQuery ( params ) , ( query ) => integration . runMetricValueQuery ( query ) , processMetricValueQueryResponse ) ; } export async function getExperimentResults ( integration , experiment , phase , metrics , activationMetric , dimension ) { const query = integration . getExperimentResultsQuery ( experiment , phase , metrics , activationMetric , dimension ) ; return getQueryDoc ( integration , query , ( ) => integration . getExperimentResults ( experiment , phase , metrics , activationMetric , dimension ) , ( rows ) => processExperimentResultsResponse ( experiment , rows ) , false ) ; } export async function getExperimentMetric ( integration , params , useCache ) { return getQueryDoc ( integration , integration . getExperimentMetricQuery ( params ) , ( query ) => integration . runExperimentMetricQuery ( query ) , ( rows ) => rows , useCache ) ; } export function processPastExperimentQueryResponse ( rows ) { return { experiments : rows . map ( ( row ) => { return { users : row . users , experiment_id : row . experiment_id , variation_id : row . variation_id , end_date : getValidDate ( row . end_date ) , start_date : getValidDate ( row . start_date ) , } ; } ) , } ; } function getVariationMap ( experiment ) { const variationMap = new Map < string , number > ( ) ; experiment . variations . forEach ( ( v , i ) => { variationMap . set ( v . key || i + "" , i ) ; } ) ; return variationMap ; } export function processExperimentResultsResponse ( experiment , rows ) { const ret = { dimensions : [ ] , unknownVariations : [ ] , } ; const variationMap = getVariationMap ( experiment ) ; const unknownVariations = new Map ( ) ; let totalUsers = 0 ; const dimensionMap = new Map < string , number > ( ) ; rows . forEach ( ( { dimension , metrics , users , variation } ) => { let i = 0 ; if ( dimensionMap . has ( dimension ) ) { i = dimensionMap . get ( dimension ) || 0 ; } else { i = ret . dimensions . length ; ret . dimensions . push ( { dimension , variations : [ ] , } ) ; dimensionMap . set ( dimension , i ) ; } const numUsers = users || 0 ; totalUsers += numUsers ; const varIndex = variationMap . get ( variation + "" ) ; if ( typeof varIndex === "undefined" || varIndex < 0 || varIndex >= experiment . variations . length ) { unknownVariations . set ( variation , numUsers ) ; return ; } const metricData = { } ; metrics . forEach ( ( { metric , ... stats } ) => { metricData [ metric ] = stats ; } ) ; ret . dimensions [ i ] . variations . push ( { variation : varIndex , users : numUsers , metrics : metricData , } ) ; } ) ; unknownVariations . forEach ( ( users , variation ) => { if ( totalUsers > 0 && users / totalUsers >= 0.02 ) { ret . unknownVariations . push ( variation ) ; } } ) ; return ret ; } export function processMetricValueQueryResponse ( rows ) { const ret = { count : 0 , mean : 0 , stddev : 0 , users : 0 } ; rows . forEach ( ( row ) => { const { date , count , mean , stddev , users , ... percentiles } = row ; if ( date ) { ret . dates = ret . dates || [ ] ; ret . dates . push ( { date , users , count , mean , stddev , } ) ; } else { ret . count = count ; ret . mean = mean ; ret . stddev = stddev ; ret . users = users ; if ( percentiles ) { Object . keys ( percentiles ) . forEach ( ( p ) => { ret . percentiles = ret . percentiles || { } ; ret . percentiles [ p . replace ( /^p/ , "" ) ] = percentiles [ p ] ; } ) ; } } } ) ; return ret ; } export async function getQueryData ( queries , organization , map ? ) { const docs = await QueryModel . find ( { organization , id : { $in : queries . map ( ( q ) => q . query ) , } , } ) ; const res = map || new Map ( ) ; docs . forEach ( ( doc ) => { const match = queries . filter ( ( q ) => q . query === doc . id ) [ 0 ] ; if ( ! match ) return ; res . set ( match . name , doc ) ; } ) ; return res ; } export async function updateQueryStatuses ( queries , organization , onUpdate , onSuccess , currentError ? ) { const byStatus = { failed : [ ] , running : [ ] , succeeded : [ ] , } ; queries . forEach ( ( q ) => { byStatus [ q . status ] . push ( q ) ; } ) ; if ( currentError || byStatus . failed . length > 0 ) { return "failed" ; } if ( byStatus . running . length === 0 ) { return "succeeded" ; } const queryMap = await getQueryData ( byStatus . running , organization ) ; let needsUpdate = false ; byStatus . running . forEach ( ( q ) => { const latest = queryMap . get ( q . name ) ; if ( ! latest ) { return ; } let status = latest . status ; if ( status === "running" && Date . now ( ) - latest . heartbeat . getTime ( ) > 150000 ) { status = "failed" ; } if ( status !== q . status ) { needsUpdate = true ; q . status = status ; } } ) ; if ( byStatus . running . filter ( ( q ) => q . status === "succeeded" ) . length === byStatus . running . length ) { await getQueryData ( byStatus . succeeded , organization , queryMap ) ; await onSuccess ( queries , queryMap ) ; return "succeeded" ; } if ( needsUpdate ) { onUpdate ( queries ) ; } return "running" ; } export async function startRun < T > ( docs , processResults ) { const queryData = new Map ( ) ; const queries = await Promise . all ( Object . keys ( docs ) . map ( async ( k ) => { const doc = await docs [ k ] ; queryData . set ( k , doc ) ; return { name : k , query : doc . id , status : doc . status , } ; } ) ) ; let result ; if ( queries . filter ( ( q ) => q . status !== "succeeded" ) . length === 0 ) { result = await processResults ( queryData ) ; } return { queries , result , } ; } export async function cancelRun < T extends InterfaceWithQueries > ( doc , organization , onDelete ) { if ( ! doc ) { throw new Error ( "Could not find document" ) ; } if ( doc . organization !== organization ) { throw new Error ( "You do not have access to this document" ) ; } if ( doc . queries . filter ( ( q ) => q . status === "running" ) . length > 0 ) { await onDelete ( ) ; } return { status : 200 , } ; } export async function getStatusEndpoint < T extends InterfaceWithQueries , R > ( doc , organization , processResults , onSave , currentError ? ) { if ( ! doc ) { throw new Error ( "Could not find document" ) ; } if ( doc . organization !== organization ) { throw new Error ( "You do not have access to this document" ) ; } const status = await updateQueryStatuses ( doc . queries , organization , async ( queries ) => { await onSave ( { queries } ) ; } , async ( queries , data ) => { let error = "" ; let results = undefined ; try { results = await processResults ( data ) ; } catch ( e ) { error = e . message ; } await onSave ( { queries } , results , error ) ; } , currentError ) ; return { status : 200 , queryStatus : status , elapsed : Math . floor ( ( Date . now ( ) - ( doc ?. runStarted ?. getTime ( ) || 0 ) ) / 1000 ) , finished : doc . queries . filter ( ( q ) => q . status === "succeeded" ) . length , total : doc . queries . length , } ; }