/*
 * 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("IS_ANALYTICS_SESSION_DATA")
@App:description("Receiving Streams from IS and executing and analyzing statistics for session")

-- Receving streams from IS
--Defining Input Stream for receiving streams from IS

define trigger TablePurgingTriggerStream at '0 0 23 * * ?';

@Async(buffer.size='32768', workers='1', batch.size.max='1024')
@source(type = "wso2event", wso2.stream.id = "org.wso2.is.analytics.stream.OverallSession:1.0.0", @map(type = 'wso2event'))
@source(type='inMemory',topic='IsAnalytics.Session.Common.Stream', @map(type='passThrough'))
define stream OverallSessionStream (
        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);

-- Defining streams to handle session data
@source(type='inMemory',topic='IsAnalytics.Session.Information.Stream', @map(type='passThrough'))
define stream HandleSessionInformation (
        meta_tenantId int,
        sessionId string,
        startTimestamp long,
        renewTimestamp long,
        terminationTimestamp long,
        endTimestamp long,
        duration long,
        isActive bool,
        username string,
        userstoreDomain string,
        remoteIp string,
        region string,
        tenantDomain string,
        serviceProvider string,
        identityProviders string,
        rememberMeFlag bool,
        userAgent string,
        userStore string,
        timestamp long);

define stream EligibleSessionFilterStream (
        meta_tenantId int,
        action int,
        _timestamp long);

-- this definition is not required, but no harm keeping it either.
define stream CountingNewTerminatedSessionsStream (
        meta_tenantId int,
        newSessionCount long,
        terminatedSessionCount long,
        net long,
        timestamp long);

@Async(buffer.size='32768', workers='2', batch.size.max='1024')
define stream SessionStatisticsPerSecond (
        meta_tenantId int,
        activeSessionCount long,
        newSessionCount long,
        terminatedSessionCount long,
        timestamp long);

--Defining Databases

@primaryKey('meta_tenantId')
@store(type='rdbms', datasource='IS_ANALYTICS_DB')
define table ActiveSessionCountTable(
        meta_tenantId int,
        activeCount long);

@primaryKey('meta_tenantId', 'sessionId')
@store(type='rdbms', datasource='IS_ANALYTICS_DB')
define table ActiveSessionsTable (
        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,
        userStore string,
        timestamp long);

@primaryKey('meta_tenantId', 'sessionId')
@Index('username','userstoreDomain','tenantDomain')
@store(type='rdbms', datasource='IS_ANALYTICS_DB')
define table SessionInformationTable(
        meta_tenantId int,
        sessionId string,
        startTime string,
        terminateTime string,
        endTime string,
        duration long,
        isActive bool,
        username string,
        userstoreDomain string,
        remoteIp string,
        region string,
        tenantDomain string,
        serviceProvider string,
        identityProviders string,
        rememberMeFlag bool,
        userAgent string,
        userStore string,
        currentTime string,
        startTimestamp long,
        renewTimestamp long,
        terminationTimestamp long,
        endTimestamp long,
        timestamp long);

-- Aggregating active, new, terminated session counts for sec-year

@store(type='rdbms', datasource='IS_ANALYTICS_DB')
@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'))
define aggregation SessionAggregation
from    SessionStatisticsPerSecond
select  meta_tenantId,
        activeSessionCount,
        sum(newSessionCount) as newSessionCount,
        sum(terminatedSessionCount) as terminatedSessionCount
group by meta_tenantId
aggregate by timestamp every sec...year;

--Queries
-- Storing active and new sessions in ActiveSessionsTable
from OverallSessionStream [action == 1 OR action == 2]#window.timeBatch(1 sec)
select  meta_tenantId,
        sessionId,
        startTimestamp,
        renewTimestamp,
        terminationTimestamp,
        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 userStore,
        _timestamp as timestamp
update or insert into
        ActiveSessionsTable
on      ActiveSessionsTable.meta_tenantId==meta_tenantId and
        ActiveSessionsTable.sessionId==sessionId;

-- capturing active and new sessions in a stream 'HandleSessionInformation'
from OverallSessionStream [action == 1 OR action == 2]
select  meta_tenantId,
        sessionId,
        startTimestamp,
        renewTimestamp,
        terminationTimestamp,
        -1L as endTimestamp,
        (_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 userStore,
        _timestamp as timestamp
insert into HandleSessionInformation;

-- capturing active and new sessions in a stream 'HandleSessionInformation'
from OverallSessionStream [action == 0]
select  meta_tenantId,
        sessionId,
        startTimestamp,
        renewTimestamp,
        terminationTimestamp,
        terminationTimestamp as endTimestamp,
        (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 userStore,
        _timestamp as timestamp
insert into HandleSessionInformation;

-- Storing session information in a table
from HandleSessionInformation
select  meta_tenantId,
        sessionId,
        time:dateFormat(startTimestamp, "MMMM dd, yyyy hh:mm:ss a zzz") as startTime,
        time:dateFormat(terminationTimestamp, "MMMM dd, yyyy hh:mm:ss a zzz") as terminateTime,
        time:dateFormat(endTimestamp, "MMMM dd, yyyy hh:mm:ss a zzz") as endTime,
        duration,
        isActive,
        username,
        userstoreDomain,
        remoteIp,
        region,
        tenantDomain,
        serviceProvider,
        identityProviders,
        rememberMeFlag,
        userAgent,
        userStore,
        time:dateFormat(timestamp, "MMMM dd, yyyy hh:mm:ss a zzz") as currentTime,
        startTimestamp,
        renewTimestamp,
        terminationTimestamp,
        endTimestamp,
        timestamp
update or insert into
        SessionInformationTable
on      SessionInformationTable.meta_tenantId==meta_tenantId and
        SessionInformationTable.sessionId==sessionId;

-- Removing inactive sessions from 'ActiveSessionsTable'
from    HandleSessionInformation[isActive == false]#window.timeBatch(1 sec)
delete  ActiveSessionsTable
on      ActiveSessionsTable.sessionId == sessionId AND
        ActiveSessionsTable.meta_tenantId == meta_tenantId;

from OverallSessionStream as o join ActiveSessionsTable as a
on      o.meta_tenantId == a.meta_tenantId
        and o.sessionId == a.sessionId
        and o.action == 0
select  o.meta_tenantId as meta_tenantId,
        o.action as action,
        o._timestamp as _timestamp
insert into
        EligibleSessionFilterStream;

from OverallSessionStream[action == 1]
select  meta_tenantId,
        action,
        _timestamp
insert into
        EligibleSessionFilterStream;

--Counting new and terminated sessions every 1 second for each meta_tenantId
from EligibleSessionFilterStream#window.externalTimeBatch(_timestamp, 1 sec, 0, 2 sec)
select  meta_tenantId,
        sum(action) as newSessionCount,
        sum(1 - action) as terminatedSessionCount,
        (2*sum(action)-count()) as net,
        _timestamp as timestamp
group by meta_tenantId
insert into
        CountingNewTerminatedSessionsStream;

-- Getting active session counts using ActiveSessionCountTable

from    CountingNewTerminatedSessionsStream as s left outer join ActiveSessionCountTable as t
on      s.meta_tenantId == t.meta_tenantId
select  s.meta_tenantId,
        (s.net  + ifThenElse(t.activeCount is null, 0L, t.activeCount) ) as activeSessionCount,
        s.newSessionCount,
        s.terminatedSessionCount,
        timestamp
group by s.meta_tenantId
insert into
        SessionStatisticsPerSecond;

-- Storing active session counts in a table 'ActiveSessionCountTable'
from    SessionStatisticsPerSecond
select  meta_tenantId,
        activeSessionCount as activeCount
update or insert into
        ActiveSessionCountTable
on      ActiveSessionCountTable.meta_tenantId==meta_tenantId;


-- Retain last one year data in the SessionInformationTable and active session table and delete the rest by sending trigger  every day

from    TablePurgingTriggerStream
select  convert(time:dateSub(triggered_time, 1, 'year'), 'long') as purgingTime
delete  SessionInformationTable
on      SessionInformationTable.timestamp < purgingTime;

from    TablePurgingTriggerStream
select  convert(time:dateSub(triggered_time, 1, 'year'), 'long') as purgingTime
delete  ActiveSessionsTable
on      ActiveSessionsTable.timestamp < purgingTime;