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

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

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

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

@Import('org.wso2.analytics.apim.apiHealthMonitorAlertStream:1.0.0')
define stream apiHealthMonitorAlertStream (api_version string, apiPublisher string, tenantDomain string, msg string, severity int, alertTimestamp long);

@Import('org.wso2.analytics.apim.abnormalResponseTimeAlertStream:1.0.0')
define stream abnormalResponseTimeAlertStream (api string, apiPublisher string, tenantDomain string, resourceTemplate string, method string, responseTime long, responsePercentile double, msg string, severity int, alertTimestamp long);

@Import('org.wso2.analytics.apim.abnormalBackendTimeAlertStream:1.0.0')
define stream abnormalBackendTimeAlertStream (api string, apiPublisher string, tenantDomain string, resourceTemplate string, method string, backendTime long, backendPercentile double, msg string, severity int, alertTimestamp long);

@Import('org.wso2.analytics.apim.tierLimitHittingAlert:1.0.0')
define stream tierLimitHittingAlert (subscriber string, apiPublisher string, api string, applicationId string, applicationName string, tenantDomain string, msg string, severity int, alertTimestamp long);

@Import('org.wso2.analytics.apim.alert.abnormalTierUsageAlert:1.1.0')
define stream abnormalTierUsageAlert (applicationId string, applicationOwner string, api_version string, tenantDomain string, msg string, severity int, alertTimestamp long);

@Import('org.wso2.analytics.apim.alert.abnormalTokenRefresh:1.0.0')
define stream abnormalTokenRefresh (userId string, tenantDomain string, scope string, timestamp long, alertTimestamp long, msg string, severity int, consumerKey string);

@Import('org.wso2.analytics.apim.alert.unusualIPAccess:1.0.0')
define stream unusualIPAccess (type string, msg string, severity int, ip string, applicationName string, applicationOwner string, userId string, tenantDomain string, requestTime long, alertTimestamp long);

@Import('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);

@Import('org.wso2.analytics.apim.abnormalRequestsPerMinAlertStream:1.0.0')
define stream abnormalRequestsPerMinAlert (api string, userId string, tenantDomain string, applicationName string, applicationOwner string,resourceTemplate string, method string, requestsPerMin long, requestsPerMinUpperPercentile double, requestsPerMinLowerPercentile double,
										   reason string, msg string, severity int, alertTimestamp long);
@Export('org.wso2.analytics.apim.emailAlertStream:1.0.0')
define stream emailAlertStream (type string, msg string, alertTimestamp string, emails string);

@from(eventtable = 'analytics.table' , table.name = 'ORG_WSO2_ANALYTICS_APIM_ALERTSTAKEHOLDERINFO' , primary.keys = 'userId,tenantDomain,isSubscriber,isPublisher,isAdmin', wait.for.indexing = 'false')
define table stakeholderInfoTable (userId string, alertTypes string, emails	string, isSubscriber bool, isPublisher bool, isAdmin bool);

@info(name = 'query1')
from abnormalRequestsPerMinAlert#window.length(1) join stakeholderInfoTable
on ((abnormalRequestsPerMinAlert.applicationOwner == stakeholderInfoTable.userId and true == stakeholderInfoTable.isSubscriber ) or true == stakeholderInfoTable.isAdmin )
select 'abnormalRequestsPerMin' as type , str:concat('Abnormal request count ' , ifThenElse(abnormalRequestsPerMinAlert.requestsPerMin > abnormalRequestsPerMinAlert.requestsPerMinUpperPercentile,'spike','drop') , ' detected during last minute by userId:',abnormalRequestsPerMinAlert.userId,' using application ', abnormalRequestsPerMinAlert.applicationName,' owned by ', abnormalRequestsPerMinAlert.applicationOwner,' for http ', abnormalRequestsPerMinAlert.method,' method of resource template ', abnormalRequestsPerMinAlert.resourceTemplate,' in api :', abnormalRequestsPerMinAlert.api,', abnormal request count:', abnormalRequestsPerMinAlert.requestsPerMin, ".") as msg, time:dateFormat(abnormalRequestsPerMinAlert.alertTimestamp,'yyyy-MM-dd HH:mm:ss') as alertTimestamp, stakeholderInfoTable.emails
having str:contains(stakeholderInfoTable.alertTypes, 'abnormalRequestsPerMin')
insert into emailAlertStream;

@info(name = 'query2')
from requestPatternChangedStream#window.length(1) join stakeholderInfoTable
on ((requestPatternChangedStream.applicationOwner == stakeholderInfoTable.userId and true == stakeholderInfoTable.isSubscriber ) or true == stakeholderInfoTable.isAdmin )
select 'RequestPatternChanged' as type , str:concat(requestPatternChangedStream.msg,  ' by user :',requestPatternChangedStream.userId,' using application : ',requestPatternChangedStream.applicationName, ' owned by: ',requestPatternChangedStream.applicationOwner, '.') as msg, time:dateFormat(requestPatternChangedStream.alertTimestamp,'yyyy-MM-dd HH:mm:ss') as alertTimestamp, stakeholderInfoTable.emails
having str:contains(stakeholderInfoTable.alertTypes, 'RequestPatternChanged')
insert into emailAlertStream;

@info(name = 'query3')
from unusualIPAccess#window.length(1) join stakeholderInfoTable
on ((unusualIPAccess.applicationOwner == stakeholderInfoTable.userId and true == stakeholderInfoTable.isSubscriber ) or true == stakeholderInfoTable.isAdmin )
select 'UnusualIPAccessAlert' as type , str:concat("A request from a ", ifThenElse(str:contains(msg, 'old'), 'old','new'), " IP (", ip, ") detected by user:" , unusualIPAccess.userId,  " using application:", applicationName, " owned by ", applicationOwner, ".") as msg, time:dateFormat(unusualIPAccess.alertTimestamp,'yyyy-MM-dd HH:mm:ss') as alertTimestamp, stakeholderInfoTable.emails
having  str:contains(stakeholderInfoTable.alertTypes, 'UnusualIPAccessAlert')
insert into emailAlertStream;

@info(name = 'query4')
from abnormalTokenRefresh#window.length(1) join stakeholderInfoTable
on ((abnormalTokenRefresh.userId == stakeholderInfoTable.userId and true == stakeholderInfoTable.isSubscriber) or true == stakeholderInfoTable.isAdmin )
select 'AbnormalRefreshAlert' as type , str:concat('Abnormal access token refresh detected by user:',abnormalTokenRefresh.userId,' using consumer key: ', consumerKey, ' at: ', time:dateFormat(timestamp,'yyyy-MM-dd HH:mm:ss'), '.') as msg, time:dateFormat(abnormalTokenRefresh.alertTimestamp,'yyyy-MM-dd HH:mm:ss') as alertTimestamp, stakeholderInfoTable.emails
having  str:contains(stakeholderInfoTable.alertTypes, 'AbnormalRefreshAlert')
insert into emailAlertStream;

@info(name = 'query5')
from abnormalTierUsageAlert#window.length(1) join stakeholderInfoTable
on ((abnormalTierUsageAlert.applicationOwner == stakeholderInfoTable.userId and true == stakeholderInfoTable.isSubscriber ) or true == stakeholderInfoTable.isAdmin )
select  'AbnormalTierUsage' as type , abnormalTierUsageAlert.msg, time:dateFormat(abnormalTierUsageAlert.alertTimestamp,'yyyy-MM-dd HH:mm:ss') as alertTimestamp ,stakeholderInfoTable.emails
having str:contains(stakeholderInfoTable.alertTypes, 'AbnormalTierUsage')
insert into emailAlertStream;

@info(name = 'query6')
from tierLimitHittingAlert#window.length(1) join stakeholderInfoTable
on ((tierLimitHittingAlert.subscriber == stakeholderInfoTable.userId and true == stakeholderInfoTable.isSubscriber ) or true == stakeholderInfoTable.isAdmin )
select 'FrequentTierHittingAlert' as type ,
ifThenElse(str:contains(tierLimitHittingAlert.msg, 'Application frequently goes beyond the allocated quota'),str:concat("The application ", tierLimitHittingAlert.applicationName, " owned by ", tierLimitHittingAlert.subscriber, " frequently goes beyond the allocated quota when accessing the ", tierLimitHittingAlert.api," API."),
str:concat(tierLimitHittingAlert.msg , " Using the ", tierLimitHittingAlert.applicationName, " application owned by ", tierLimitHittingAlert.subscriber, ".")) as msg, time:dateFormat(tierLimitHittingAlert.alertTimestamp,'yyyy-MM-dd HH:mm:ss') as alertTimestamp, stakeholderInfoTable.emails
having  str:contains(stakeholderInfoTable.alertTypes, 'FrequentTierHittingAlert')
insert into emailAlertStream;

@info(name = 'query7')
from abnormalResponseTimeAlertStream#window.length(1) join stakeholderInfoTable
on ((abnormalResponseTimeAlertStream.apiPublisher == stakeholderInfoTable.userId and true == stakeholderInfoTable.isPublisher ) or true == stakeholderInfoTable.isAdmin )
select 'abnormalResponseTime' as type , str:concat('Abnormal response time detected for http ',abnormalResponseTimeAlertStream.method,' method of resource template:',abnormalResponseTimeAlertStream.resourceTemplate,' in api:',abnormalResponseTimeAlertStream.api,' of tenant domain:',abnormalResponseTimeAlertStream.tenantDomain,', threshold value:',abnormalResponseTimeAlertStream.responsePercentile, 'ms.') as msg, time:dateFormat(abnormalResponseTimeAlertStream.alertTimestamp,'yyyy-MM-dd HH:mm:ss') as alertTimestamp, stakeholderInfoTable.emails
having str:contains(stakeholderInfoTable.alertTypes, 'abnormalResponseTime')
insert into emailAlertStream;

@info(name = 'query8')
from abnormalBackendTimeAlertStream#window.length(1) join stakeholderInfoTable
on ((abnormalBackendTimeAlertStream.apiPublisher == stakeholderInfoTable.userId and true == stakeholderInfoTable.isPublisher ) or true == stakeholderInfoTable.isAdmin )
select 'abnormalBackendTime' as type , str:concat('Abnormal backend response time detected for http ', abnormalBackendTimeAlertStream.method,' method of resource template:', abnormalBackendTimeAlertStream.resourceTemplate,' in api:',abnormalBackendTimeAlertStream.api,' of tenant domain:',abnormalBackendTimeAlertStream.tenantDomain,', threshold value:',abnormalBackendTimeAlertStream.backendPercentile, 'ms.') as msg, time:dateFormat(abnormalBackendTimeAlertStream.alertTimestamp,'yyyy-MM-dd HH:mm:ss') as alertTimestamp, stakeholderInfoTable.emails
having  str:contains(stakeholderInfoTable.alertTypes, 'abnormalBackendTime')
insert into emailAlertStream;

@info(name = 'query9')
from apiHealthMonitorAlertStream#window.length(1) join stakeholderInfoTable
on ((apiHealthMonitorAlertStream.apiPublisher == stakeholderInfoTable.userId and true == stakeholderInfoTable.isPublisher ) or true == stakeholderInfoTable.isAdmin )
select 'healthAvailabilityPerMin' as type , str:concat('API:', api_version, '-', msg) as msg, time:dateFormat(apiHealthMonitorAlertStream.alertTimestamp,'yyyy-MM-dd HH:mm:ss') as alertTimestamp, stakeholderInfoTable.emails
having str:contains(stakeholderInfoTable.alertTypes, 'healthAvailabilityPerMin')
insert into emailAlertStream;