/*
 * Copyright (c) 2018 WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
 *
 * WSO2 Inc. licenses this file to you under the Apache License,
 * Version 2.0 (the "License"); you may not use this file except
 * in compliance with the License.
 * You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

@App:name('EI_Analytics_StatApp')
@App:description('SiddhiApp for EI Analytics')

-- Streams

-- temporary stream to store ESBStat data
define stream PreProcessedESBStatStream (componentId string, componentName string, componentType string, duration long, faultCount int, startTime long, entryPoint string, metaTenantId int);

-- temporary stream to store MediatorStat data
define stream PreProcessedMediatorStatStream (metaTenantId int, entryPoint string, entryPointHashcode string, componentId string, hashCode string, componentName string, componentType string, duration long, faultCount int, startTime long);

-- maps Flow Entry into FlowEntryEventStream
@source(type = 'wso2event', wso2.stream.id = 'org.wso2.esb.analytics.stream.FlowEntry:1.0.0',
	@map(type = 'wso2event'))
define stream FlowEntryEventStream (meta_compressed bool, meta_tenantId int, messageId string, flowData string);

-- maps Config entry into ConfigEntryEventStream
@source(type = 'wso2event', wso2.stream.id = 'org.wso2.esb.analytics.stream.ConfigEntry:1.0.0',
	@map(type = 'wso2event'))
define stream ConfigEntryEventStream (meta_tenantId int, hashcode string, entryName string, configData string);

-- defines the DecompressedEventStream, which stores decompressed FlowEntryEventStream data
define stream DecompressedEventStream (metaTenantId int, messageFlowId string, host string, hashCode string, componentName string, componentType string, componentIndex int, componentId string, startTime long, endTime long, duration long, beforePayload string, afterPayload string, contextPropertyMap string, transportPropertyMap string, children string, entryPoint string, entryPointHashcode string, faultCount int, eventTimestamp long);

-- defines the TableInsertAsyncStream, which inserts events asynchronously into ESBEventTable
@Async(buffer.size = '8192', workers = '4', batch.size.max = '2000')
define stream ESBEventTableInsertAsyncStream (metaTenantId int, messageFlowId string, host string, hashCode string,
componentName string, componentType string, componentIndex int, componentId string, startTime long, endTime long, duration long, beforePayload string, afterPayload string, contextPropertyMap string, transportPropertyMap string, children string, entryPoint string, entryPointHashcode string, faultCount int, eventTimestamp long);

-- Tables

-- table that stores ESBEventStream data
@store(type = 'rdbms', datasource = 'EI_ANALYTICS', field.length="contextPropertyMap:5000,beforePayload:5000,afterPayload:5000,transportPropertyMap:5000" )
@Index('metaTenantId','messageFlowId')
define table ESBEventTable (metaTenantId int, messageFlowId string, host string, hashCode string, componentName string, componentType string, componentIndex int, componentId string, startTime long, endTime long, duration long, beforePayload string, afterPayload string, contextPropertyMap string, transportPropertyMap string, children string, entryPoint string, entryPointHashcode string, faultCount int, eventTimestamp long);

-- table that stores ConfigEntryEventStream data
@store(type = 'rdbms', datasource = 'EI_ANALYTICS', field.length="configData:6000")
@PrimaryKey('hashcode')
@Index('metaTenantId','entryName')
define table ConfigEntryTable (metaTenantId int, hashcode string, entryName string, configData string, eventTimestamp long);

-- table that stores component name and type for the searchbox-widget
@PrimaryKey('componentId')
@Index('componentType')
define table ComponentNameInMemoryTable (componentId string, componentName string, componentType string);

@store(type = 'rdbms', datasource = 'EI_ANALYTICS')
@PrimaryKey('componentId')
@Index('componentType')
define table ComponentNameTable (componentId string, componentName string, componentType string);

-- Aggregations

-- aggregates PreProcessedMediatorStatStream data every minute to month
@store(type = 'rdbms', datasource = 'EI_ANALYTICS')
@info(name = 'MediatorStat')
@purge(enable='true', interval='60 min', @retentionPeriod(sec='1 day', min='72 hours', hours='90 days', days='1 year', months='2 years', years='3 years'))
@PartitionById
define aggregation MediatorStatAgg
from PreProcessedMediatorStatStream
 select metaTenantId, entryPoint, entryPointHashcode, componentId, hashCode, componentName, componentType, sum(duration) as totalDuration ,avg(duration) as avgDuration, min(duration) as minDuration, max(duration) as maxDuration, count() as noOfInvocation, sum(faultCount) as faultCount, startTime
	group by metaTenantId, componentId, componentName, componentType, entryPoint, entryPointHashcode, hashCode
	aggregate by startTime every sec...years;

-- aggregates PreProcessedESBStatStream data every minute to month
@store(type = 'rdbms', datasource = 'EI_ANALYTICS')
@purge(enable='true', interval='60 min', @retentionPeriod(sec='1 day', min='72 hours', hours='90 days', days='1 year', months='2 years', years='3 years'))
@info(name = 'ESBStat')
@PartitionById
define aggregation ESBStatAgg
from PreProcessedESBStatStream
 select componentId, componentName, componentType, sum(duration) as totalDuration, avg(duration) as avgDuration, min(duration) as minDuration, max(duration) as maxDuration, count() as noOfInvocation, sum(faultCount) as faultCount, entryPoint, metaTenantId, startTime as eventTimestamp
	group by metaTenantId, componentId, componentName, componentType, entryPoint
	aggregate by startTime every sec...years;

--Triggers
define trigger TablePurgingTriggerStream at '0 0 23 * * ?';
define trigger AppStartTriggerStream at 'start';

-- Queries

-- decompress compressed-FlowEntryEventStream and store data in DecompressedEventStream
from FlowEntryEventStream#esbAnalytics:decompress(meta_compressed, meta_tenantId, messageId, flowData)
select metaTenantId, messageFlowId, host, hashCode, componentName, str:lower(componentType) as componentType, componentIndex, componentId, startTime, endTime, duration, beforePayload, afterPayload, contextPropertyMap, transportPropertyMap, children, entryPoint, entryPointHashcode, faultCount, _timestamp as eventTimestamp
insert current events into DecompressedEventStream;

-- store ConfigEntryEventStream data into ConfigEntryTable
from ConfigEntryEventStream
select meta_tenantId as metaTenantId, hashcode, entryName, configData, eventTimestamp() as eventTimestamp
update or insert into ConfigEntryTable on ConfigEntryTable.hashcode == hashcode;

-- add all the componentIds, componentNames and componentTypes to ComponentNameTable
from AppStartTriggerStream join ComponentNameTable
select componentId, componentName, componentType
insert into ComponentNameInMemoryTable;

from DecompressedEventStream[not((ComponentNameInMemoryTable.componentId == componentId) in ComponentNameInMemoryTable)]
select componentId, componentName, componentType
insert into ComponentNameTableStream;

from ComponentNameTableStream
select *
update or insert into ComponentNameInMemoryTable on ComponentNameInMemoryTable.componentId == componentId;

from ComponentNameTableStream
select *
update or insert into ComponentNameTable on ComponentNameTable.componentId == componentId;

-- if DecompressedEventStream has beforePayload or transportPropertyMap or contextPropertyMap, add them to into ESBEventTable through an async stream
from DecompressedEventStream[not(beforePayload is null) or not(transportPropertyMap is null) or not(contextPropertyMap is null)]
select metaTenantId, messageFlowId, host, hashCode, componentName, componentType, componentIndex, componentId, startTime, endTime, duration, beforePayload, afterPayload, contextPropertyMap, transportPropertyMap, children, entryPoint, entryPointHashcode, faultCount, eventTimestamp
insert current events into ESBEventTableInsertAsyncStream;

from ESBEventTableInsertAsyncStream
select *
insert current events into ESBEventTable;

-- if DecompressedEventStream's componentType is ProxyService or API or InboundEndPoint, add it into PreProcessedESBStatStream
from DecompressedEventStream[componentType == "proxy service" or componentType == "api" or componentType == "inbound endpoint"]
select componentId, componentName, componentType, duration, ifThenElse(faultCount > 0, 1, 0) as faultCount, startTime, entryPoint, metaTenantId
insert current events into PreProcessedESBStatStream;

-- insert DecompressedEventStream data into PreProcessedMediatorStatStream
from DecompressedEventStream
select metaTenantId, entryPoint, entryPointHashcode, componentId, hashCode, componentName, componentType, duration, ifThenElse(faultCount > 0, 1, 0) as faultCount, startTime
insert current events into PreProcessedMediatorStatStream;

from    TablePurgingTriggerStream
select  convert(time:dateSub(triggered_time, 2, 'week'), 'long') as purgingTime
delete  ConfigEntryTable
on      ConfigEntryTable.eventTimestamp < purgingTime;

from    TablePurgingTriggerStream
select  convert(time:dateSub(triggered_time, 2, 'week'), 'long') as purgingTime
delete  ESBEventTable
on      ESBEventTable.eventTimestamp < purgingTime;
