import Agenda , { Job } from "agenda" ; import { ExperimentModel } from "../models/ExperimentModel" ; import { getDataSourceById } from "../models/DataSourceModel" ; import { isEmailEnabled , sendExperimentChangesEmail } from "../services/email" ; import { createSnapshot , getExperimentWatchers , getLatestSnapshot , } from "../services/experiments" ; import { getConfidenceLevelsForOrg } from "../services/organizations" ; import pino from "pino" ; import { ExperimentSnapshotDocument , ExperimentSnapshotModel , } from "../models/ExperimentSnapshotModel" ; import { ExperimentInterface } from "../../types/experiment" ; import { getStatusEndpoint } from "../services/queries" ; import { getMetricById } from "../models/MetricModel" ; import { EXPERIMENT_REFRESH_FREQUENCY } from "../util/secrets" ; import { analyzeExperimentResults } from "../services/stats" ; import { getReportVariations } from "../services/reports" ; import { findOrganizationById } from "../models/OrganizationModel" ; const UPDATE_EVERY = EXPERIMENT_REFRESH_FREQUENCY * 60 * 60 * 1000 ; const QUEUE_EXPERIMENT_UPDATES = "queueExperimentUpdates" ; const UPDATE_SINGLE_EXP = "updateSingleExperiment" ; type UpdateSingleExpJob = Job < { experimentId : string ; } > ; const parentLogger = pino ( ) ; export default async function ( agenda ) { agenda . define ( QUEUE_EXPERIMENT_UPDATES , async ( ) => { const ids = await legacyQueueExperimentUpdates ( ) ; const experimentIds = ( await ExperimentModel . find ( { datasource : { $exists : true , $ne : "" , } , status : "running" , autoSnapshots : true , nextSnapshotAttempt : { $exists : true , $lte : new Date ( ) , } , id : { $nin : ids , } , } , { id : true , } ) . limit ( 100 ) . sort ( { nextSnapshotAttempt : 1 , } ) ) . map ( ( e ) => e . id ) ; for ( let i = 0 ; i < experimentIds . length ; i ++ ) { await queueExerimentUpdate ( experimentIds [ i ] ) ; } } ) ; agenda . define ( UPDATE_SINGLE_EXP , { lockLifetime : 30 * 60 * 1000 } , updateSingleExperiment ) ; await startUpdateJob ( ) ; async function legacyQueueExperimentUpdates ( ) { const latestDate = new Date ( Date . now ( ) - UPDATE_EVERY ) ; const experimentIds = ( await ExperimentModel . find ( { datasource : { $exists : true , $ne : "" , } , status : "running" , autoSnapshots : true , nextSnapshotAttempt : { $exists : false , } , lastSnapshotAttempt : { $lte : latestDate , } , } , { id : true , } ) . limit ( 100 ) . sort ( { lastSnapshotAttempt : 1 , } ) ) . map ( ( e ) => e . id ) ; for ( let i = 0 ; i < experimentIds . length ; i ++ ) { await queueExerimentUpdate ( experimentIds [ i ] ) ; } return experimentIds ; } async function startUpdateJob ( ) { const updateResultsJob = agenda . create ( QUEUE_EXPERIMENT_UPDATES , { } ) ; updateResultsJob . unique ( { } ) ; updateResultsJob . repeatEvery ( "10 minutes" ) ; await updateResultsJob . save ( ) ; } async function queueExerimentUpdate ( experimentId ) { const job = agenda . create ( UPDATE_SINGLE_EXP , { experimentId , } ) as UpdateSingleExpJob ; job . unique ( { experimentId , } ) ; job . schedule ( new Date ( ) ) ; await job . save ( ) ; } } async function updateSingleExperiment ( job ) { const experimentId = job . attrs . data ?. experimentId ; if ( ! experimentId ) return ; const logger = parentLogger . child ( { cron : "updateSingleExperiment" , experimentId , } ) ; const experiment = await ExperimentModel . findOne ( { id : experimentId , } ) ; if ( ! experiment ) return ; let lastSnapshot ; let currentSnapshot ; try { logger . info ( "Start Refreshing Results" ) ; const datasource = await getDataSourceById ( experiment . datasource || "" , experiment . organization ) ; if ( ! datasource ) return ; lastSnapshot = await getLatestSnapshot ( experiment . id , experiment . phases . length - 1 ) ; const organization = await findOrganizationById ( experiment . organization ) ; if ( ! organization ) return ; if ( organization ?. settings ?. updateSchedule ?. type === "never" ) return ; currentSnapshot = await createSnapshot ( experiment , experiment . phases . length - 1 , organization , null , false ) ; await new Promise < void > ( ( resolve , reject ) => { const check = async ( ) => { const phase = experiment . phases [ experiment . phases . length - 1 ] ; if ( ! phase ) { reject ( "Invalid phase" ) ; return ; } const res = await getStatusEndpoint ( currentSnapshot , currentSnapshot . organization , ( queryData ) => { return analyzeExperimentResults ( experiment . organization , getReportVariations ( experiment , phase ) , undefined , queryData ) ; } , async ( updates , results , error ) => { await ExperimentSnapshotModel . updateOne ( { id : currentSnapshot . id } , { $set : { ... updates , unknownVariations : results ?. unknownVariations || [ ] , multipleExposures : results ?. multipleExposures || 0 , results : results ?. dimensions || currentSnapshot . results , error , } , } ) ; } , currentSnapshot . error ) ; if ( res . queryStatus === "succeeded" ) { resolve ( ) ; return ; } if ( res . queryStatus === "failed" ) { reject ( "Queries failed to run" ) ; return ; } setTimeout ( check , 10000 ) ; } ; setTimeout ( check , 2000 ) ; } ) ; logger . info ( "Success" ) ; await sendSignificanceEmail ( experiment , lastSnapshot , currentSnapshot ) ; } catch ( e ) { logger . error ( "Failure - " + e . message ) ; try { experiment . autoSnapshots = false ; experiment . markModified ( "autoSnapshots" ) ; await experiment . save ( ) ; } catch ( e ) { logger . error ( "Failed to turn off autoSnapshots - " + e . message ) ; } } } async function sendSignificanceEmail ( experiment , lastSnapshot , currentSnapshot ) { const logger = parentLogger . child ( { cron : "sendSignificanceEmail" , experimentId : experiment . id , } ) ; if ( ! isEmailEnabled ( ) ) { return ; } if ( ! currentSnapshot ?. results ?. [ 0 ] ?. variations ) { return ; } try { const { ciUpper , ciLower } = await getConfidenceLevelsForOrg ( experiment . organization ) ; const experimentChanges = [ ] ; for ( let i = 1 ; i < currentSnapshot . results [ 0 ] . variations . length ; i ++ ) { const curVar = currentSnapshot . results ?. [ 0 ] ?. variations ?. [ i ] ; const lastVar = lastSnapshot . results ?. [ 0 ] ?. variations ?. [ i ] ; for ( const m in curVar . metrics ) { const curMetric = curVar ?. metrics ?. [ m ] ; const lastMetric = lastVar ?. metrics ?. [ m ] ; if ( lastMetric ?. chanceToWin && curMetric ?. chanceToWin && curMetric ?. value > 150 ) { if ( curMetric . chanceToWin > ciUpper && lastMetric . chanceToWin < ciUpper ) { experimentChanges . push ( "The metric " + getMetricById ( m , experiment . organization ) + " for variation " + experiment . variations [ i ] . name + " has reached a " + ( curMetric . chanceToWin * 100 ) . toFixed ( 1 ) + "% chance to beat baseline" ) ; } else if ( curMetric . chanceToWin < ciLower && lastMetric . chanceToWin > ciLower ) { experimentChanges . push ( "The metric " + getMetricById ( m , experiment . organization ) + " for variation " + experiment . variations [ i ] . name + " has dropped to a " + ( curMetric . chanceToWin * 100 ) . toFixed ( 1 ) + " chance to beat the baseline" ) ; } } } } if ( experimentChanges . length ) { logger . info ( "Significant change - detected " + experimentChanges . length + " significant changes" ) ; const watchers = await getExperimentWatchers ( experiment . id ) ; const userIds = watchers . map ( ( w ) => w . userId ) ; await sendExperimentChangesEmail ( userIds , experiment . id , experiment . name , experimentChanges ) ; } } catch ( e ) { logger . error ( e . message ) ; } }