Ghost/ghost/email-analytics-service/lib/email-analytics-service.js

const EventProcessingResult = require('./event-processing-result');
const debug = require('@tryghost/debug')('services:email-analytics');
const logging = require('@tryghost/logging');

/**
 * @typedef {import('@tryghost/email-service').EmailEventProcessor} EmailEventProcessor
 */

module.exports = class EmailAnalyticsService {
    config;
    settings;
    queries;
    eventProcessor;
    providers;

    /**
     * @param {object} dependencies
     * @param {EmailEventProcessor} dependencies.eventProcessor
     */
    constructor({config, settings, queries, eventProcessor, providers}) {
        this.config = config;
        this.settings = settings;
        this.queries = queries;
        this.eventProcessor = eventProcessor;
        this.providers = providers;
    }

    async fetchAll() {
        const result = new EventProcessingResult();

        const shouldFetchStats = await this.queries.shouldFetchStats();
        if (!shouldFetchStats) {
            debug('fetchAll: skipping - fetch requirements not met');
            return result;
        }

        const startFetch = new Date();
        debug('fetchAll: starting');
        for (const [, provider] of Object.entries(this.providers)) {
            const providerResults = await provider.fetchAll(this.processEventBatch.bind(this));
            result.merge(providerResults);
        }
        debug(`fetchAll: finished (${Date.now() - startFetch}ms)`);

        return result;
    }

    async fetchLatest({maxEvents = Infinity} = {}) {
        const result = new EventProcessingResult();

        const shouldFetchStats = await this.queries.shouldFetchStats();
        if (!shouldFetchStats) {
            debug('fetchLatest: skipping - fetch requirements not met');
            return result;
        }

        const lastTimestamp = await this.queries.getLastSeenEventTimestamp();

        const startFetch = new Date();
        debug('fetchLatest: starting');
        providersLoop:
        for (const [, provider] of Object.entries(this.providers)) {
            const providerResults = await provider.fetchLatest(lastTimestamp, this.processEventBatch.bind(this), {maxEvents});
            result.merge(providerResults);

            if (result.totalEvents >= maxEvents) {
                break providersLoop;
            }
        }
        debug(`fetchLatest: finished in ${Date.now() - startFetch}ms. Fetched ${result.totalEvents} events`);

        return result;
    }

    async processEventBatch(events) {
        const result = new EventProcessingResult();

        for (const event of events) {
            const batchResult = await this.processEvent(event);
            result.merge(batchResult);
        }

        return result;
    }

    /**
     *
     * @param {{id: string, type: any; severity: any; recipientEmail: any; emailId: any; providerId: string; timestamp: Date; error: {code: number; message: string; enhandedCode: string|number} | null}} event
     * @returns {Promise<EventProcessingResult>}
     */
    async processEvent(event) {
        if (event.type === 'delivered') {
            const recipient = await this.eventProcessor.handleDelivered({emailId: event.emailId, providerId: event.providerId, email: event.recipientEmail}, event.timestamp);

            if (recipient) {
                return new EventProcessingResult({
                    delivered: 1,
                    emailIds: [recipient.emailId],
                    memberIds: [recipient.memberId]
                });
            }

            return new EventProcessingResult({unprocessable: 1});
        }

        if (event.type === 'opened') {
            const recipient = await this.eventProcessor.handleOpened({emailId: event.emailId, providerId: event.providerId, email: event.recipientEmail}, event.timestamp);

            if (recipient) {
                return new EventProcessingResult({
                    opened: 1,
                    emailIds: [recipient.emailId],
                    memberIds: [recipient.memberId]
                });
            }

            return new EventProcessingResult({unprocessable: 1});
        }

        if (event.type === 'failed') {
            if (event.severity === 'permanent') {
                const recipient = await this.eventProcessor.handlePermanentFailed({emailId: event.emailId, providerId: event.providerId, email: event.recipientEmail}, {id: event.id, timestamp: event.timestamp, error: event.error});

                if (recipient) {
                    return new EventProcessingResult({
                        permanentFailed: 1,
                        emailIds: [recipient.emailId],
                        memberIds: [recipient.memberId]
                    });
                }

                return new EventProcessingResult({unprocessable: 1});
            } else {
                const recipient = await this.eventProcessor.handleTemporaryFailed({emailId: event.emailId, providerId: event.providerId, email: event.recipientEmail}, {id: event.id, timestamp: event.timestamp, error: event.error});

                if (recipient) {
                    return new EventProcessingResult({
                        temporaryFailed: 1,
                        emailIds: [recipient.emailId],
                        memberIds: [recipient.memberId]
                    });
                }

                return new EventProcessingResult({unprocessable: 1});
            }
        }

        if (event.type === 'unsubscribed') {
            const recipient = await this.eventProcessor.handleUnsubscribed({emailId: event.emailId, providerId: event.providerId, email: event.recipientEmail}, event.timestamp);

            if (recipient) {
                return new EventProcessingResult({
                    unsubscribed: 1,
                    emailIds: [recipient.emailId],
                    memberIds: [recipient.memberId]
                });
            }

            return new EventProcessingResult({unprocessable: 1});
        }

        if (event.type === 'complained') {
            const recipient = await this.eventProcessor.handleComplained({emailId: event.emailId, providerId: event.providerId, email: event.recipientEmail}, event.timestamp);

            if (recipient) {
                return new EventProcessingResult({
                    complained: 1,
                    emailIds: [recipient.emailId],
                    memberIds: [recipient.memberId]
                });
            }

            return new EventProcessingResult({unprocessable: 1});
        }

        return new EventProcessingResult({unhandled: 1});
    }

    async aggregateStats({emailIds = [], memberIds = []}) {
        logging.info(`Aggregating email analytics for ${emailIds.length} emails`);
        for (const emailId of emailIds) {
            await this.aggregateEmailStats(emailId);
        }

        logging.info(`Aggregating email analytics for ${memberIds.length} members`);
        for (const memberId of memberIds) {
            await this.aggregateMemberStats(memberId);
        }
    }

    async aggregateEmailStats(emailId) {
        return this.queries.aggregateEmailStats(emailId);
    }

    async aggregateMemberStats(memberId) {
        return this.queries.aggregateMemberStats(memberId);
    }
};