/* Enter a unique ExecutionPlan */
@Plan:name('IsAnalytics-ExecutionPlan-SessionData')

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

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

@async(bufferSize='1024')
@Import('org.wso2.is.analytics.stream.OverallSession:1.0.0')
define stream SessionStream (meta_tenantId int, sessionId string, startTimestamp long, renewTimestamp long, terminationTimestamp long, action int, username string, userstoreDomain string, remoteIp string, region string, tenantDomain string, serviceProvider string, identityProviders string, rememberMeFlag bool, userAgent string, _timestamp long);

@Export('org.wso2.is.analytics.stream.SessionStatPerMinute:1.0.0')
define stream SessionStatPerMinute (meta_tenantId int, bucketId long, bucketStart long, bucketEnd long, year int, month int, day int, hour int, minute int, activeSessionCount long, newSessionCount long, terminatedSessionCount long);

@from(eventtable = 'analytics.table', table.name = 'ORG_WSO2_IS_ANALYTICS_STREAM_ACTIVESESSIONS', primary.keys = 'meta_tenantId, sessionId' , indices = 'meta_tenantId, sessionId , startTimestamp, renewTimestamp, terminationTimestamp, year, month, day, hour, minute, action, username, userstoreDomain, remoteIp, region, tenantDomain, rememberMeFlag, userAgent',  wait.for.indexing = 'true', merge.schema = 'false')
define table activeSessionsTable (meta_tenantId int, sessionId string, startTimestamp long, renewTimestamp long, terminationTimestamp long, year int, month int, day int, hour int, minute int, action int, username string, userstoreDomain string, remoteIp string, region string, tenantDomain string, serviceProvider string, identityProviders string, rememberMeFlag bool, userAgent string, usernameWithTenantDomainAndUserstoreDomain string);

@from(eventtable = 'analytics.table' , table.name = 'ORG_WSO2_IS_ANALYTICS_STREAM_ACTIVECOUNT', primary.keys = 'meta_tenantId, id', indices = 'meta_tenantId, id, activeCount', wait.for.indexing = 'true', merge.schema = 'false')
define table activeSessionCountTable(meta_tenantId int, id string, activeCount long);

@Export('org.wso2.is.analytics.stream.SessionInfo:1.0.0')
define stream SessionStreamInfo (meta_tenantId int, sessionId string, startTimestamp long, renewTimestamp long, terminationTimestamp long, endTimestamp long, year int, month int, day int, hour int, minute int, duration long, isActive bool, username string, userstoreDomain string, remoteIp string, region string, tenantDomain string, serviceProvider string, identityProviders string, rememberMeFlag bool, userAgent string, usernameWithTenantDomainAndUserstoreDomain string);

from SessionStream
select meta_tenantId, sessionId, startTimestamp, renewTimestamp, terminationTimestamp, action, username, userstoreDomain, remoteIp, region, tenantDomain, serviceProvider, identityProviders, rememberMeFlag, userAgent, time:timestampInMilliseconds() as _timestamp
insert into IntermediateSessionStream;

from IntermediateSessionStream [action == 1 OR action == 2]#window.timeBatch(1 sec)
select meta_tenantId, sessionId, startTimestamp, renewTimestamp, terminationTimestamp, time:extract(startTimestamp, 'year') as year, time:extract(startTimestamp, 'month') as month, time:extract(startTimestamp, 'day') as day, time:extract(startTimestamp, 'hour') as hour, time:extract(startTimestamp, 'minute') as minute, action, username, userstoreDomain, remoteIp, region, tenantDomain, serviceProvider, identityProviders, rememberMeFlag, userAgent, ifThenElse(userstoreDomain is null OR userstoreDomain == "" OR userstoreDomain == "NOT_AVAILABLE" OR tenantDomain is null OR tenantDomain == "" OR tenantDomain == "NOT_AVAILABLE", username, str:concat(userstoreDomain, '/', username, '@', tenantDomain)) as usernameWithTenantDomainAndUserstoreDomain
insert into activeSessionsTable;

from IntermediateSessionStream [action == 1 OR action == 2]
select meta_tenantId, sessionId, startTimestamp, renewTimestamp, terminationTimestamp, convert(-1, 'long') as endTimestamp, time:extract(startTimestamp, 'year') as year, time:extract(startTimestamp, 'month') as month, time:extract(startTimestamp, 'day') as day, time:extract(startTimestamp, 'hour') as hour, time:extract(startTimestamp, 'minute') as minute, (_timestamp - startTimestamp) as duration, true as isActive, username, userstoreDomain, remoteIp, region, tenantDomain, serviceProvider, identityProviders, rememberMeFlag, userAgent, ifThenElse(userstoreDomain is null OR userstoreDomain == "" OR userstoreDomain == "NOT_AVAILABLE" OR tenantDomain is null OR tenantDomain == "" OR tenantDomain == "NOT_AVAILABLE", username, str:concat(userstoreDomain, '/', username, '@', tenantDomain)) as usernameWithTenantDomainAndUserstoreDomain
insert into SessionStreamInfo;

from IntermediateSessionStream [action == 0]
select meta_tenantId, sessionId, startTimestamp, renewTimestamp, terminationTimestamp, terminationTimestamp as endTimestamp, time:extract(startTimestamp, 'year') as year, time:extract(startTimestamp, 'month') as month, time:extract(startTimestamp, 'day') as day, time:extract(startTimestamp, 'hour') as hour, time:extract(startTimestamp, 'minute') as minute, (terminationTimestamp - startTimestamp) as duration,  false as isActive, username, userstoreDomain, remoteIp, region, tenantDomain, serviceProvider, identityProviders, rememberMeFlag, userAgent, ifThenElse(userstoreDomain is null OR userstoreDomain == "" OR userstoreDomain == "NOT_AVAILABLE" OR tenantDomain is null OR tenantDomain == "" OR tenantDomain == "NOT_AVAILABLE", username, str:concat(userstoreDomain, '/', username, '@', tenantDomain)) as usernameWithTenantDomainAndUserstoreDomain
insert into SessionStreamInfo;

from SessionStreamInfo[isActive == false]#window.timeBatch(1 sec)
delete activeSessionsTable
on activeSessionsTable.sessionId == sessionId AND activeSessionsTable.meta_tenantId == meta_tenantId;

from IntermediateSessionStream[action == 1 OR action == 0]#window.externalTimeBatch(_timestamp, 1 min, 0, 2 min)
select meta_tenantId, (_timestamp - _timestamp%60000) as bucketId, sum(action) as newSessionCount, sum(1 - action) as terminatedSessionCount, (2*sum(action)-count()) as net, 'count' as id
group by meta_tenantId
insert into BucketStream;

from BucketStream#window.length(1) as s left outer join activeSessionCountTable as t
on s.id == t.id AND s.meta_tenantId == t.meta_tenantId
select s.meta_tenantId, s.bucketId, s.bucketId as bucketStart, (s.bucketId + 60000) as bucketEnd, time:extract(bucketId, 'year') as year, time:extract(bucketId, 'month') as month, time:extract(bucketId, 'day') as day, time:extract(bucketId, 'hour') as hour, time:extract(bucketId, 'minute') as minute,  (s.net  + ifThenElse(t.activeCount is null, convert(0, 'long'), t.activeCount) ) as activeSessionCount, s.newSessionCount, s.terminatedSessionCount
group by s.meta_tenantId
insert into SessionStatPerMinute;

from SessionStatPerMinute#window.length(1)
select meta_tenantId, 'count' as id, activeSessionCount as activeCount
insert into activeSessionCountTable;
