Skip to content

Tested email analytics database queries #29890

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
EvanHahn wants to merge 4 commits into from
Open

Tested email analytics database queries #29890

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
713585f
Make email analytics queries helper a class
EvanHahn Aug 11, 2026
54d4b08
`Queries` should take a database reference instead of importing it
EvanHahn Aug 11, 2026
5701baf
Test `Queries.prototype.getJobData`
EvanHahn Aug 11, 2026
ce3793e
Test remaining Queries methods
EvanHahn Aug 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion ghost/core/core/server/services/email-analytics/email-analytics-service-wrapper.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ const logging = require('@tryghost/logging');
const metrics = require('@tryghost/metrics');
const config = require('../../../shared/config');
const domainEvents = require('@tryghost/domain-events');
/** @import {Queries} from './lib/queries' */
/** @import {PrometheusClient} from '@tryghost/prometheus-metrics' */
/** @import {BatchEventProcessor} from './batch-event-processor' */
/** @import {JobNames, CursorSeed, EmailAnalyticsFetchResult} from './email-analytics-service' */
Expand All @@ -28,6 +29,7 @@ class EmailAnalyticsServiceWrapper {
/**
* @param {object} options
* @param {Parameters<typeof domainEvents.subscribe>[0]} options.event
* @param {Queries} options.queries
* @param {string[]} options.mailgunTags
* @param {JobNames} options.jobNames
* @param {CursorSeed} options.cursorSeed
Expand All @@ -36,6 +38,7 @@ class EmailAnalyticsServiceWrapper {
*/
init({
event,
queries,
mailgunTags,
jobNames,
cursorSeed,
Expand All @@ -49,7 +52,6 @@ class EmailAnalyticsServiceWrapper {
const {EmailAnalyticsService} = require('./email-analytics-service');
const {fetchMailgunEvents} = require('./fetch-mailgun-events');
const settings = require('../../../shared/settings-cache');
const {queries} = require('./lib/queries');

this.service = new EmailAnalyticsService({
fetchEvents: (options) => fetchMailgunEvents({...options, config, settings, tags: mailgunTags}),
Expand Down
6 changes: 5 additions & 1 deletion ghost/core/core/server/services/email-analytics/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@ import {EmailRecipientFailure, EmailSpamComplaintEvent, Email} from '../../model
import domainEvents from '@tryghost/domain-events';
// @ts-expect-error This module lacks type definitions.
import prometheusClient from '../../../shared/prometheus-client';
import {queries} from './lib/queries';
import {Queries} from './lib/queries';
import {StartEmailAnalyticsJobEvent} from './events/start-email-analytics-job-event';
import {StartAutomationEmailAnalyticsJobEvent} from './events/start-automation-email-analytics-job-event';
import {AUTOMATION_EMAIL_TAG} from '../member-welcome-emails/constants';
Expand All @@ -34,6 +34,8 @@ export const automations = new EmailAnalyticsServiceWrapper({
});

export const init = () => {
const queries = new Queries(db.knex);

const newsletterEmailEventProcessor = new EmailEventProcessor({
domainEvents,
db,
Expand All @@ -58,6 +60,7 @@ export const init = () => {

newsletters.init({
event: StartEmailAnalyticsJobEvent,
queries,
mailgunTags: newsletterMailgunTags,
jobNames: {
latestNonOpened: 'email-analytics-latest-others',
Expand Down Expand Up @@ -86,6 +89,7 @@ export const init = () => {

automations.init({
event: StartAutomationEmailAnalyticsJobEvent,
queries,
mailgunTags: [AUTOMATION_EMAIL_TAG],
jobNames: {
latestNonOpened: 'email-analytics-automation-latest-others',
Expand Down
100 changes: 56 additions & 44 deletions ghost/core/core/server/services/email-analytics/lib/queries.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,9 +1,8 @@
import _ from 'lodash';
import debugFactory from '@tryghost/debug';
// @ts-expect-error This module lacks type definitions.
import db from '../../../data/db';
import logging from '@tryghost/logging';
import ObjectID from 'bson-objectid';
import type {Knex} from 'knex';
import type {JsonObject} from 'type-fest';

const debug = debugFactory('services:email-analytics');
Expand Down Expand Up @@ -31,8 +30,8 @@ type JobData = {
* Creates a job in the jobs table if it does not already exist.
* @param jobName - The name of the job to create.
*/
async function createJobIfNotExists(jobName: EmailAnalyticsJobName): Promise<void> {
await db.knex('jobs').insert({
async function createJobIfNotExists(knex: Knex, jobName: EmailAnalyticsJobName): Promise<void> {
await knex('jobs').insert({
id: new ObjectID().toHexString(),
name: jobName,
started_at: new Date(),
Expand Down Expand Up @@ -67,7 +66,13 @@ function parseJobMetadata(rawMetadata: unknown): JobMetadata {
};
}

export const queries = {
export class Queries {
#knex: Knex;

constructor(knex: Knex) {
this.#knex = knex;
}

/**
* Retrieves the timestamp of the last seen event for the specified email analytics events.
* @param jobName - The name of the job to update.
Expand All @@ -80,6 +85,8 @@ export const queries = {
events: EmailAnalyticsEvent[],
cursorSeed: CursorSeed
): Promise<Date | null> {
const knex = this.#knex;

const startDate = new Date();

let timestamps: (Date | string | null)[] = [];
Expand All @@ -97,13 +104,13 @@ export const queries = {
if (!columnName) {
continue;
}
const row = await db.knex(cursorSeed.tableName)
.select(db.knex.raw('MAX(??) as maxTimestamp', [columnName]))
const row = await knex(cursorSeed.tableName)
.select(knex.raw('MAX(??) as maxTimestamp', [columnName]))
.first();
timestamps.push(row.maxTimestamp);
}

await createJobIfNotExists(jobName);
await createJobIfNotExists(knex, jobName);
}

// Convert string dates to Date objects for SQLite compatibility
Expand All @@ -118,15 +125,15 @@ export const queries = {
debug(`getLastEventTimestamp: finished in ${Date.now() - startDate.getTime()}ms`);

return lastSeenEventTimestamp;
},
}

/**
* Retrieves the job data for the specified job name.
* @param jobName - The name of the job to retrieve data for.
* @returns The job data, or null if no job data is found.
*/
async getJobData(jobName: EmailAnalyticsJobName): Promise<JobData | null> {
const row = await db.knex('jobs')
const row = await this.#knex('jobs')
.select('finished_at', 'started_at', 'metadata')
.where('name', jobName)
.first();
Expand All @@ -135,7 +142,7 @@ export const queries = {
started_at: row.started_at,
metadata: parseJobMetadata(row.metadata)
} : null;
},
}

/**
* Retrieves the timestamp of the last job run for the specified job name.
Expand All @@ -145,7 +152,7 @@ export const queries = {
async getLastJobRunTimestamp(jobName: EmailAnalyticsJobName): Promise<Date | null> {
const jobData = await this.getJobData(jobName);
return jobData ? jobData.finished_at || jobData.started_at : null;
},
}

/**
* Sets the timestamp of the last seen event for the specified email analytics events.
Expand All @@ -162,9 +169,9 @@ export const queries = {
debug(`Setting ${field} timestamp for job ${jobName} to ${date}`);
const updateField = field === 'finished' ? 'finished_at' : 'started_at';
const status = field === 'finished' ? 'finished' : 'started';
const result = await db.knex('jobs').update({[updateField]: date, updated_at: new Date(), status: status}).where('name', jobName);
const result = await this.#knex('jobs').update({[updateField]: date, updated_at: new Date(), status: status}).where('name', jobName);
if (result === 0) {
await db.knex('jobs').insert({
await this.#knex('jobs').insert({
id: new ObjectID().toHexString(),
name: jobName,
[updateField]: date.toISOString(), // force to iso string for sqlite
Expand All @@ -176,17 +183,17 @@ export const queries = {
const message = err instanceof Error ? err.message : String(err);
debug(`Error setting ${field} timestamp for job ${jobName}: ${message}`);
}
},
}

/**
* Retrieves and parses the metadata JSON for the specified job.
* @param jobName - The name of the job.
* @returns The parsed metadata object, or null.
*/
async getJobMetadata(jobName: EmailAnalyticsJobName): Promise<JobMetadata | null> {
const row = await db.knex('jobs').select('metadata').where('name', jobName).first();
const row = await this.#knex('jobs').select('metadata').where('name', jobName).first();
return row ? parseJobMetadata(row.metadata) : null;
},
}

/**
* Writes metadata JSON for the specified job.
Expand All @@ -196,7 +203,7 @@ export const queries = {
async setJobMetadata(jobName: EmailAnalyticsJobName, metadata: JobMetadata | null): Promise<void> {
try {
const value = metadata ? JSON.stringify(metadata) : null;
await db.knex.transaction(async (trx: typeof db.knex) => {
await this.#knex.transaction(async (trx: Knex.Transaction) => {
const result = await trx('jobs').update({metadata: value, updated_at: new Date()}).where('name', jobName);
if (result === 0 && metadata) {
await trx('jobs').insert({
Expand All @@ -212,7 +219,7 @@ export const queries = {
const message = err instanceof Error ? err.message : String(err);
logging.error(`Error setting metadata for job ${jobName}: ${message}`);
}
},
}

/**
* Sets the status of the specified email analytics job.
Expand All @@ -225,15 +232,15 @@ export const queries = {
async setJobStatus(jobName: EmailAnalyticsJobName, status: 'started' | 'finished' | 'failed'): Promise<void> {
debug(`Setting status for job ${jobName} to ${status}`);
try {
const result = await db.knex('jobs')
const result = await this.#knex('jobs')
.update({
status: status,
updated_at: new Date()
})
.where('name', jobName);

if (result === 0) {
await db.knex('jobs').insert({
await this.#knex('jobs').insert({
id: new ObjectID().toHexString(),
name: jobName,
status: status,
Expand All @@ -246,63 +253,70 @@ export const queries = {
debug(`Error setting status for job ${jobName}: ${message}`);
throw err;
}
},
}

async aggregateEmailStats(emailId: string, updateOpenedCount: boolean): Promise<void> {
const [deliveredCount] = await db.knex('email_recipients').count('id as count').whereRaw('email_id = ? AND delivered_at IS NOT NULL', [emailId]);
const [failedCount] = await db.knex('email_recipients').count('id as count').whereRaw('email_id = ? AND failed_at IS NOT NULL', [emailId]);
const [deliveredCount] = await this.#knex('email_recipients').count('id as count').whereRaw('email_id = ? AND delivered_at IS NOT NULL', [emailId]);
const [failedCount] = await this.#knex('email_recipients').count('id as count').whereRaw('email_id = ? AND failed_at IS NOT NULL', [emailId]);

const updateData: Record<string, string | number> = {
delivered_count: deliveredCount.count,
failed_count: failedCount.count
};

if (updateOpenedCount) {
const [openedCount] = await db.knex('email_recipients').count('id as count').whereRaw('email_id = ? AND opened_at IS NOT NULL', [emailId]);
const [openedCount] = await this.#knex('email_recipients').count('id as count').whereRaw('email_id = ? AND opened_at IS NOT NULL', [emailId]);
updateData.opened_count = openedCount.count;
}

await db.knex('emails').update(updateData).where('id', emailId);
},
await this.#knex('emails').update(updateData).where('id', emailId);
}

async aggregateMemberStats(memberId: string): Promise<void> {
const {trackedEmailCount} = await db.knex('email_recipients')
.select(db.knex.raw('COUNT(email_recipients.id) as trackedEmailCount'))
const {trackedEmailCount} = await this.#knex('email_recipients')
.select(this.#knex.raw('COUNT(email_recipients.id) as trackedEmailCount'))
.leftJoin('emails', 'email_recipients.email_id', 'emails.id')
.where('email_recipients.member_id', memberId)
.where('emails.track_opens', true)
.first() || {};

const [emailCount] = await db.knex('email_recipients').count('id as count').whereRaw('member_id = ?', [memberId]);
const [emailOpenedCount] = await db.knex('email_recipients').count('id as count').whereRaw('member_id = ? AND opened_at IS NOT NULL', [memberId]);
const emailCountResult = await this.#knex('email_recipients')
.count('id as count')
.whereRaw('member_id = ?', [memberId])
.first();
const emailOpenedCountResult = await this.#knex('email_recipients')
.count('id as count')
.whereRaw('member_id = ? AND opened_at IS NOT NULL', [memberId])
.first();
const emailCount = Number(emailCountResult?.count || 0);
const emailOpenedCount = Number(emailOpenedCountResult?.count || 0);

const updateQuery: Record<string, string | number> = {
email_count: emailCount.count,
email_opened_count: emailOpenedCount.count
email_count: emailCount,
email_opened_count: emailOpenedCount
};

if (trackedEmailCount >= MIN_EMAIL_COUNT_FOR_OPEN_RATE) {
updateQuery.email_open_rate = Math.round(emailOpenedCount.count / trackedEmailCount * 100);
updateQuery.email_open_rate = Math.round(emailOpenedCount / trackedEmailCount * 100);
}

await db.knex('members')
await this.#knex('members')
.update(updateQuery)
.where('id', memberId);
},
}

async aggregateMemberStatsBatch(memberIds: string[]): Promise<void> {
if (!memberIds || memberIds.length === 0) {
return;
}

// Batch query to get stats for all members at once
const stats = await db.knex('email_recipients')
const stats = await this.#knex('email_recipients')
.leftJoin('emails', 'emails.id', 'email_recipients.email_id')
.select(
'email_recipients.member_id',
db.knex.raw('COUNT(email_recipients.id) as email_count'),
db.knex.raw('SUM(CASE WHEN email_recipients.opened_at IS NOT NULL THEN 1 ELSE 0 END) as email_opened_count'),
db.knex.raw('SUM(CASE WHEN emails.track_opens = 1 THEN 1 ELSE 0 END) as tracked_count')
this.#knex.raw('COUNT(email_recipients.id) as email_count'),
this.#knex.raw('SUM(CASE WHEN email_recipients.opened_at IS NOT NULL THEN 1 ELSE 0 END) as email_opened_count'),
this.#knex.raw('SUM(CASE WHEN emails.track_opens = 1 THEN 1 ELSE 0 END) as tracked_count')
)
.whereIn('email_recipients.member_id', memberIds)
.groupBy('email_recipients.member_id');
Expand Down Expand Up @@ -364,7 +378,7 @@ export const queries = {
];

// Execute batched update with CASE statements
await db.knex.raw(`
await this.#knex.raw(`
UPDATE members
SET
email_count = CASE id ${emailCountCases.join(' ')} END,
Expand All @@ -374,5 +388,3 @@ export const queries = {
`, bindings);
}
};

export type Queries = typeof queries;
6 changes: 3 additions & 3 deletions ghost/core/test/integration/services/email-analytics/automation-email-analytics.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@ const {agentProvider} = require('../../../utils/e2e-framework');
const testUtils = require('../../../utils');
const MailgunClient = require('../../../../core/server/services/lib/mailgun-client');
const {AUTOMATION_EMAIL_TAG, DEFAULT_EMAIL_DESIGN_SETTING_SLUG} = require('../../../../core/server/services/member-welcome-emails/constants');
const {queries} = require('../../../../core/server/services/email-analytics/lib/queries');
const {Queries} = require('../../../../core/server/services/email-analytics/lib/queries');
const emailAnalytics = require('../../../../core/server/services/email-analytics');
const automationsApi = require('../../../../core/server/services/automations/automations-api');

Expand All @@ -28,9 +28,9 @@ describe('Automation email analytics', function () {
let emailDesignSettingId;

beforeAll(async function () {
sinon.stub(queries, 'getLastEventTimestamp').resolves(new Date(2000, 0, 1));
sinon.stub(Queries.prototype, 'getLastEventTimestamp').resolves(new Date(2000, 0, 1));
// Same reason, for the cursor fetchMissing starts from.
sinon.stub(queries, 'getLastJobRunTimestamp').resolves(new Date(2000, 0, 1));
sinon.stub(Queries.prototype, 'getLastJobRunTimestamp').resolves(new Date(2000, 0, 1));

await agentProvider.getAdminAPIAgent();

Expand Down
4 changes: 2 additions & 2 deletions ghost/core/test/integration/services/email-service/newsletter-email-event-storage.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,8 @@ processingModes.forEach(({name, batchProcessing}) => {
configUtils.set('emailAnalytics:batchProcessing', batchProcessing);

// Stub queries before boot
const {queries} = require('../../../../core/server/services/email-analytics/lib/queries');
sinon.stub(queries, 'getLastEventTimestamp').callsFake(async function () {
const {Queries} = require('../../../../core/server/services/email-analytics/lib/queries');
sinon.stub(Queries.prototype, 'getLastEventTimestamp').callsFake(async function () {
// This is required because otherwise the last event timestamp will be now, and that is too close to NOW to start fetching new events
return new Date(2000, 0, 1);
});
Expand Down
4 changes: 2 additions & 2 deletions ghost/core/test/integration/services/mailgun-email-suppression-list.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,8 +18,8 @@ describe('MailgunEmailSuppressionList', function () {
// fetched, so no suppression is created. The suite previously passed only
// by free-riding on an earlier file's timestamp state, which breaks under
// per-file isolation. Mirrors newsletter-email-event-storage.test.js.
const {queries} = require('../../../core/server/services/email-analytics/lib/queries');
sinon.stub(queries, 'getLastEventTimestamp').callsFake(async function () {
const {Queries} = require('../../../core/server/services/email-analytics/lib/queries');
sinon.stub(Queries.prototype, 'getLastEventTimestamp').callsFake(async function () {
return new Date(2000, 0, 1);
});

Expand Down
Loading
Loading