@Plan:name('APIMAnalytics-RequestPatternChangeDetection-APIRequestPatternChangeAnalysisMetric-realtime1')


@Import('org.wso2.analytics.apim.transitionStream:1.0.0')
define stream metricStream (startState string, endState string, consumerKey string, applicationName string, applicationOwner string, 
									 userId string, tenantDomain string, requestTimestamp string);

@Export('org.wso2.analytics.apim.requestPatternChangedStream:1.0.0')
define stream requestPatternChangedStream (userId string, applicationName string, applicationOwner string, tenantDomain string, msg string, severity int, alertTimestamp long);

@Export('org.wso2.analytics.apim.allApimAlertsStream:1.0.0')
define stream all_apim_alerts_stream (type string, tenantDomain string, msg string, severity int, alertTimestamp long);


@from(eventtable = 'analytics.table' , table.name = 'org_wso2_analytics_apim_MARKOVMODELTABLE', primary.keys='startState,endState, consumerKey', indices = 'startState,endState, probability, consumerKey', caching = 'true', cache.timeout.seconds='300', wait.for.indexing = 'false' )
define table MarkovModelTable (consumerKey string, startState string, endState string, probability double);

@from(eventtable = 'analytics.table' , table.name = 'org_wso2_analytics_apim_REQUESTCOUNTTABLE', primary.keys='consumerKey', indices = 'consumerKey,count', caching = 'true', cache.timeout.seconds='60', wait.for.indexing = 'false')
define table RequestCountPerConsumerKeyTable(consumerKey string, count long);

@from(eventtable = 'analytics.table' , table.name = 'ORG_WSO2_ANALYTICS_APIM_REQUESTPATTERNALERTSUMMARYTABLE', primary.keys='applicationName, applicationOwner, userId, msg' , indices = 'applicationName, applicationOwner, userId, msg, lastlastAlertedTimestamp', wait.for.indexing = 'false')
define table RequestPatternAlertSummaryTable(applicationName string, applicationOwner string, userId string, msg string, lastAlertedTimestamp long);

/****************************************************** Calculate Miss Probability *******************************************************************************/
@info(name = 'query1')
from metricStream as ms left outer join MarkovModelTable
on (ms.startState == MarkovModelTable.startState and ms.endState == MarkovModelTable.endState and ms.consumerKey == MarkovModelTable.consumerKey)
select ms.userId, ms.requestTimestamp, ms.startState ,ms.endState, ifThenElse(MarkovModelTable.probability is null, 1.0, 1- MarkovModelTable.probability) as MissProbability, ms.consumerKey,  ms.applicationName, ms.applicationOwner, ms.tenantDomain
insert into metricValueStreamTemp;

/****************************************************** fetch request count for consumerkey *******************************************************************************/
@info(name = 'query2')
from metricValueStreamTemp as m join RequestCountPerConsumerKeyTable
on (consumerKey == RequestCountPerConsumerKeyTable.consumerKey)
select userId, requestTimestamp, startState, endState, MissProbability, m.consumerKey, m.applicationName, m.applicationOwner , m.tenantDomain,  count
insert into metricValueStream;

/****************************************************** Compare Normalize Miss Probability with Threshold *******************************************************/
@info(name = 'query5')
from metricValueStream [ MissProbability > 0.95 and count >= 500]
select userId, requestTimestamp,  str:concat(startState,' to ',endState) as transition, MissProbability as miss_probability, consumerKey, applicationName, applicationOwner, tenantDomain
insert into IntermediateFraudStream;

@info(name = 'query6')
from IntermediateFraudStream
select userId, applicationName, applicationOwner, tenantDomain, str:concat('Abnormal request pattern detected.' ,' Suspicious API transition is: ',transition) as msg , transition, (time:timestampInMilliseconds()) as alertTimestamp
insert into requestPatternChangedStreamTemp;

@info(name = 'query7')
from requestPatternChangedStreamTemp as r left outer join RequestPatternAlertSummaryTable
on (applicationName == RequestPatternAlertSummaryTable.applicationName   and applicationOwner == RequestPatternAlertSummaryTable.applicationOwner and userId == RequestPatternAlertSummaryTable.userId and msg == RequestPatternAlertSummaryTable.msg)
select r.applicationName, r.applicationOwner, r.tenantDomain, r.userId, r.msg, r.transition, alertTimestamp , ifThenElse(lastAlertedTimestamp is null, 0l , lastAlertedTimestamp) as lastAlertedTimestamp
insert into requestPatternChangedStreamModified;

@info(name = 'query8')
from requestPatternChangedStreamModified [(alertTimestamp - lastAlertedTimestamp) > 30*60*1000]
select userId, applicationName, applicationOwner, tenantDomain, msg, 1 as severity, alertTimestamp
insert into requestPatternChangedStream;

@info(name = 'query9')
from requestPatternChangedStream[(applicationName == RequestPatternAlertSummaryTable.applicationName   and applicationOwner == RequestPatternAlertSummaryTable.applicationOwner and userId == RequestPatternAlertSummaryTable.userId and msg == RequestPatternAlertSummaryTable.msg ) in RequestPatternAlertSummaryTable]
select  applicationName, applicationOwner,userId, msg, alertTimestamp as lastAlertedTimestamp
update RequestPatternAlertSummaryTable
on (applicationName == RequestPatternAlertSummaryTable.applicationName and  applicationOwner == RequestPatternAlertSummaryTable.applicationOwner and userId == RequestPatternAlertSummaryTable.userId and msg == RequestPatternAlertSummaryTable.msg);

@info(name = 'query10')
from requestPatternChangedStream[not ((applicationName == RequestPatternAlertSummaryTable.applicationName and applicationOwner == RequestPatternAlertSummaryTable.applicationOwner and userId == RequestPatternAlertSummaryTable.userId and msg == RequestPatternAlertSummaryTable.msg) in RequestPatternAlertSummaryTable)]
select  applicationName, applicationOwner,userId, msg,  alertTimestamp as lastAlertedTimestamp
insert into RequestPatternAlertSummaryTable;

@info(name = 'query11')
from requestPatternChangedStreamModified[(alertTimestamp - lastAlertedTimestamp) > 30*60*1000]
select 'RequestPatternChanged' as type, tenantDomain, str:concat('Abnormal request pattern detected by user :',userId,' using application : ',applicationName, ' owned by: ',applicationOwner, ' suspicious API transition is: ',transition, '.') as msg, 1 as severity, (time:timestampInMilliseconds()) as alertTimestamp
insert into all_apim_alerts_stream;