EventCache.ts

import { Session, SessionManager } from '../sessions/SessionManager';
import { v4 } from 'uuid';
import { MetaData } from '../events/meta-data';
import { Config } from '../orchestration/Orchestration';
import { PageAttributes, PageManager } from '../sessions/PageManager';
import {
    AppMonitorDetails,
    UserDetails,
    RumEvent,
    ParsedRumEvent
} from '../dispatch/dataplane';
import EventBus, { Topic } from '../event-bus/EventBus';
import { RecordEvent } from '../plugins/types';
import { SESSION_START_EVENT_TYPE } from '../plugins/utils/constant';
import { InternalLogger } from '../utils/InternalLogger';

const webClientVersion = '1.26.0';

/**
 * A cache which stores events generated by telemetry plugins.
 *
 * The event cache stores meta data and events until they are dispatched to the
 * data plane. The event cache removes the oldest event once the cache is full
 * and a new event is added.
 */
export class EventCache {
    private appMonitorDetails: AppMonitorDetails;
    private config: Config;

    private events: RumEvent[] = [];
    private candidates = new Map<string, RumEvent>();

    private sessionManager: SessionManager;
    private pageManager: PageManager;

    private enabled: boolean;
    private installationMethod: string;

    // Enable config.debug mode
    private droppedEvent = 0; // tracks dropped event count due to insufficient `eventCache`
    private sessionLimitExceeded = 0; // tracks dropped event count due to insufficieent `sessionEventLimit`

    /**
     * @param applicationDetails Application identity and version.
     * @param batchLimit The maximum number of events that will be returned in a batch.
     * @param eventCacheSize  The maximum number of events the cache can contain before dropping events.
     * @param sessionManager  The sessionManager returns user id, session id and handles session timeout.
     * @param pageManager The pageManager returns page id.
     */
    constructor(
        applicationDetails: AppMonitorDetails,
        config: Config,
        private eventBus = new EventBus<Topic>()
    ) {
        this.appMonitorDetails = applicationDetails;
        this.config = config;
        this.enabled = true;
        this.pageManager = new PageManager(config, this.recordEvent);
        this.sessionManager = new SessionManager(
            applicationDetails,
            config,
            this.recordEvent,
            this.pageManager
        );
        this.installationMethod = config.client;
    }

    /**
     * The event cache will record new events or new meta data.
     */
    public enable(): void {
        this.enabled = true;
    }

    /**
     * The event cache will not record new events or new meta data. Events and
     * meta data which are already in the cache will still be accessible.
     */
    public disable(): void {
        this.enabled = false;
    }

    /**
     * Update the current page interaction for the session.
     */
    public recordPageView = (payload: string | PageAttributes) => {
        if (this.isCurrentUrlAllowed()) {
            this.pageManager.recordPageView(payload);
        }
    };

    /**
     * Returns true if the session is sampled, false otherwise.
     */
    public isSessionSampled(): boolean {
        return this.sessionManager.isSampled();
    }

    /**
     * Add an event to the cache and reset the session timer.
     *
     * If the session is being recorded, the event will be recorded.
     * If the session is not being recorded, the event will not be recorded.
     *
     * @param type The event schema.
     * @param eventData The event data.
     */
    public recordEvent: RecordEvent = (type: string, eventData: object) => {
        if (!this.enabled) {
            return;
        }
        if (this.isCurrentUrlAllowed()) {
            if (type !== SESSION_START_EVENT_TYPE) {
                // Only refresh session if not session start event
                // to avoid recursive loop.
                this.sessionManager.getSession();
            }
            if (this.sessionManager.canRecord()) {
                this.sessionManager.incrementSessionEventCount();
                this.addRecordToCache(type, eventData);
            } else if (this.config.debug) {
                this.sessionLimitExceeded++;
            }
        }
    };

    /**
     * Adds a candidate to the cache and reset session timer
     *
     * @param eventType The event schema.
     * @param eventData The event data.
     */
    public recordCandidate: RecordEvent = (
        eventType: string,
        eventData: object
    ) => {
        const session = this.sessionManager.getSession();
        if (!this.enabled || !this.isCurrentUrlAllowed() || !session.record) {
            return;
        }

        const [event] = this.createEvent(eventType, eventData);

        // Update candidate if exists
        if (this.candidates.has(eventType)) {
            this.candidates.set(eventType, event);
            return;
        }

        // Record new candidate only if limits have not been reached
        if (
            this.candidates.size < this.config.candidatesCacheSize &&
            !this.sessionManager.isLimitExceeded()
        ) {
            this.candidates.set(eventType, event);
            this.sessionManager.incrementSessionEventCount();
        }
    };

    /**
     * Returns the current session (1) if a session exists and (2) if the
     * current URL is allowed. Returns undefined otherwise.
     */
    public getSession = (): Session | undefined => {
        if (this.isCurrentUrlAllowed()) {
            return this.sessionManager.getSession();
        }
        return undefined;
    };

    /**
     * Returns true if there are one or more events in the cache.
     */
    public hasEvents(): boolean {
        // For debug mode
        if (this.config.debug && this.sessionLimitExceeded > 0) {
            const total = this.events.length + this.sessionLimitExceeded;
            if (this.sessionManager.isSampled()) {
                InternalLogger.warn(
                    `Dropped ${
                        this.sessionLimitExceeded
                    } of ${total} recently observed events (${(
                        (this.sessionLimitExceeded / total) *
                        100
                    ).toFixed(
                        2
                    )}%) because the session limit has exceeded. Consider increasing sessionEventLimit (currently ${
                        this.config.sessionEventLimit
                    }) to ${total} or more to avoid data loss.`
                );
            } else {
                InternalLogger.warn(
                    `Dropped ${
                        this.sessionLimitExceeded
                    } of ${total} recently observed events (${(
                        (this.sessionLimitExceeded / total) *
                        100
                    ).toFixed(
                        2
                    )}%) because current session is not sampled. Consider increasing sessionSampleRate (currently ${
                        this.config.sessionSampleRate
                    }) to capture more user sessions.`
                );
            }
            // reset
            this.sessionLimitExceeded = 0;
        }

        return this.events.length !== 0;
    }

    /**
     * Returns true if there are one or more event candidates in the cache.
     */
    public hasCandidates() {
        return this.candidates.size !== 0;
    }

    /**
     * Removes and returns the next batch of events.
     */
    public getEventBatch(flushCandidates = false): RumEvent[] {
        if (this.config.debug && this.droppedEvent > 0) {
            const total = this.events.length + this.droppedEvent;
            InternalLogger.warn(
                `Dropped ${
                    this.droppedEvent
                } of ${total} recently observed events (${(
                    (this.droppedEvent / total) *
                    100
                ).toFixed(
                    2
                )}%) due to insufficient in-memory queue. Increase eventCacheSize to ${total} (currently ${
                    this.config.eventCacheSize
                }) to avoid data loss.`
            );
            // reset
            this.droppedEvent = 0;
        }

        let batch: RumEvent[] = [];

        // Prioritize candidates in the next event batch
        if (flushCandidates && this.hasCandidates()) {
            // Pull all candidates if they fit in the batch
            if (this.candidates.size <= this.config.batchLimit) {
                batch = Array.from(this.candidates.values());
                this.candidates.clear();
            } else {
                // Pull candidates in FIFO order until batch limit is reached
                let i = 0;
                for (const key of Array.from(this.candidates.keys())) {
                    if (i++ >= this.config.batchLimit) {
                        break;
                    }
                    const event = this.candidates.get(key);
                    if (event) {
                        batch.push(event);
                        this.candidates.delete(key);
                    }
                }
            }
        }

        // Use remaining capacity for regular events.
        if (this.events.length) {
            if (this.events.length <= this.config.batchLimit - batch.length) {
                batch.push(...this.events);
                this.events = [];
            } else {
                // Dispatch the front of the array and retain the back of the array.
                batch.push(
                    ...this.events.splice(
                        0,
                        this.config.batchLimit - batch.length
                    )
                );
            }
        }

        return batch;
    }

    /**
     * Returns an object containing the AppMonitor ID and application version.
     */
    public getAppMonitorDetails(): AppMonitorDetails {
        return this.appMonitorDetails;
    }

    /**
     * Returns an object containing the session ID and user ID.
     */
    public getUserDetails(): UserDetails {
        return {
            userId: this.sessionManager.getUserId(),
            sessionId: this.sessionManager.getSession().sessionId
        };
    }

    /**
     * Set custom session attributes to add them to all event metadata.
     *
     * @param payload object containing custom attribute data in the form of key, value pairs
     */
    public addSessionAttributes(sessionAttributes: {
        [k: string]: string | number | boolean;
    }): void {
        this.sessionManager.addSessionAttributes(sessionAttributes);
    }

    /**
     * Add an event to the cache.
     *
     * @param type The event schema.
     */
    private addRecordToCache = (type: string, eventData: object) => {
        if (!this.enabled) {
            return;
        }

        if (this.events.length >= this.config.eventCacheSize) {
            if (this.config.debug) {
                this.droppedEvent += 1;
            }
            // Drop newest event and keep the older ones
            // 1. Older events tend to be more relevant, such as session start
            //    or performance entries that are attributed to web vitals
            // 2. Dropping an old event requires linear time
            return;
        }

        const [event, parsedEvent] = this.createEvent(type, eventData);
        this.eventBus.dispatch(Topic.EVENT, parsedEvent);
        this.events.push(event);
    };

    /** Creates a RumEvent and a ParsedRumEvent from a type and details. */
    private createEvent = (
        type: string,
        details: object
    ): [RumEvent, ParsedRumEvent] => {
        // The data plane service model (i.e., LogEvents) does not adhere to the
        // RUM agent data model, where sessions and pages are first class
        // objects with their own attribute sets. Instead, we store session
        // attributes and page attributes together as 'meta data'.
        const metadata = {
            ...this.sessionManager.getAttributes(),
            ...this.pageManager.getAttributes(),
            version: '1.0.0',
            'aws:client': this.installationMethod,
            'aws:clientVersion': webClientVersion
        } as MetaData;

        const partialEvent = {
            id: v4(),
            timestamp: new Date(),
            type
        };

        return [
            {
                ...partialEvent,
                details: JSON.stringify(details),
                metadata: JSON.stringify(metadata)
            } as RumEvent,
            {
                ...partialEvent,
                details,
                metadata
            } as ParsedRumEvent
        ];
    };

    /**
     * Returns {@code true} if the current url matches one of the allowedPages
     * and does not match any of the deniedPages; returns {@code false}
     * otherwise.
     */
    private isCurrentUrlAllowed() {
        const location = document.location.toString();
        const exclude = this.config.pagesToExclude.some((re) =>
            re.test(location)
        );

        const include = this.config.pagesToInclude.some((re) =>
            re.test(location)
        );

        return include && !exclude;
    }
}