/* Enter a unique ExecutionPlan */
@Plan:name('APIMAnalytics-MarkovStateClassifier-MarkovStateClassifier-realtime1')

/* Enter a unique description for ExecutionPlan */
-- @Plan:description('ExecutionPlan')

/* define streams/tables and write queries here ... */


@Import('org.wso2.analytics.apim.stateStream:1.0.0')
define stream stateStream (context string, method string, resourceTemplate string, tenantDomain string, state string, userId string, requestTimestamp string, consumerKey string, applicationName string, applicationOwner string);

@Import('org.wso2.apimgt.statistics.requestsPerMinPerKeyStream:1.0.0')
define stream requestsPerMinPerKeyStream (consumerKey string, requestsPerMinPerKey long);
   
@Export('org.wso2.analytics.apim.transitionStream:1.0.0')
define stream stateTransitionStream (startState string, endState string, consumerKey string, applicationName string, applicationOwner string, 
									 userId string, tenantDomain string, requestTimestamp string);

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


@info(name = 'query2')
from requestsPerMinPerKeyStream as r left outer join RequestCountPerConsumerKeyTable
on (r.consumerKey == RequestCountPerConsumerKeyTable.consumerKey)
select r.consumerKey , ifThenElse(count is null,r.requestsPerMinPerKey,count+r.requestsPerMinPerKey) as count
insert into requestCountPerConsumerKeyTableStream;

@info(name = 'query3')
from requestCountPerConsumerKeyTableStream#window.timeBatch(1 min)
select consumerKey,count
insert into RequestCountPerConsumerKeyTable;


@info(name = 'query4')
from stateStream#amAnalytics:buildTransitions(consumerKey, userId, state, applicationName, applicationOwner, tenantDomain, requestTimestamp, 10000)
select startState, state as endState, consumerKey, applicationName, applicationOwner, userId, tenantDomain, requestTimestamp 
insert into stateTransitionStream;