Added new email batch sending service (#15865)

fixes https://github.com/TryGhost/Team/issues/2284

New batch sending flow (still WIP). Logs the sent emails instead of actually sending them. Unit tests are coming in later commits.

ghost/core/core/server/api/endpoints/emails.js

@@ -2,6 +2,8 @@ const models = require('../../models');
 const tpl = require('@tryghost/tpl');
 const errors = require('@tryghost/errors');
 const megaService = require('../../services/mega');
+const emailService = require('../../services/email-service');
+const labs = require('../../../shared/labs');
 
 const messages = {
     emailNotFound: 'Email not found.',
@@ -57,23 +59,27 @@ module.exports = {
             'id'
         ],
         permissions: true,
-        query(frame) {
-            return models.Email.findOne(frame.data, frame.options)
-                .then(async (model) => {
-                    if (!model) {
-                        throw new errors.NotFoundError({
-                            message: tpl(messages.emailNotFound)
-                        });
-                    }
+        // (complexity removed with new labs flag)
+        // eslint-disable-next-line ghost/ghost-custom/max-api-complexity
+        async query(frame) {
+            if (labs.isSet('emailStability')) {
+                return await emailService.controller.retryFailedEmail(frame);
+            }
 
-                    if (model.get('status') !== 'failed') {
-                        throw new errors.IncorrectUsageError({
-                            message: tpl(messages.retryNotAllowed)
-                        });
-                    }
-
-                    return await megaService.mega.retryFailedEmail(model);
+            const model = await models.Email.findOne(frame.data, frame.options);
+            if (!model) {
+                throw new errors.NotFoundError({
+                    message: tpl(messages.emailNotFound)
                 });
+            }
+
+            if (model.get('status') !== 'failed') {
+                throw new errors.IncorrectUsageError({
+                    message: tpl(messages.retryNotAllowed)
+                });
+            }
+
+            return await megaService.mega.retryFailedEmail(model);
         }
     }
 };

ghost/core/core/server/services/email-service/index.js

@@ -1,16 +1,3 @@
-class EmailServiceWrapper {
-    init() {
-        const {EmailService, EmailController} = require('@tryghost/email-service');
-        const {Post, Newsletter} = require('../../models');
-
-        this.service = new EmailService({});
-        this.controller = new EmailController(this.service, {
-            models: {
-                Post,
-                Newsletter
-            }
-        });
-    }
-}
+const EmailServiceWrapper = require('./wrapper');
 
 module.exports = new EmailServiceWrapper();

ghost/core/core/server/services/email-service/wrapper.js

@@ -0,0 +1,64 @@
+const logging = require('@tryghost/logging');
+const ObjectID = require('bson-objectid').default;
+
+class EmailServiceWrapper {
+    init() {
+        const {EmailService, EmailController, EmailRenderer, SendingService, BatchSendingService, EmailSegmenter} = require('@tryghost/email-service');
+        const {Post, Newsletter, Email, EmailBatch, EmailRecipient, Member} = require('../../models');
+        const settingsCache = require('../../../shared/settings-cache');
+        const jobsService = require('../jobs');
+        const membersService = require('../members');
+        const db = require('../../data/db');
+        const membersRepository = membersService.api.members;
+        const limitService = require('../limits');
+
+        const emailRenderer = new EmailRenderer();
+        const sendingService = new SendingService({
+            emailProvider: {
+                send: ({plaintext, subject, from, replyTo, recipients}) => {
+                    logging.info(`Sending email\nSubject: ${subject}\nFrom: ${from}\nReplyTo: ${replyTo}\nRecipients: ${recipients.length}\n\n${plaintext}`);
+                    return Promise.resolve({id: 'fake_provider_id_' + ObjectID().toHexString()});
+                }
+            },
+            emailRenderer
+        });
+
+        const emailSegmenter = new EmailSegmenter({
+            membersRepository
+        });
+        const batchSendingService = new BatchSendingService({
+            sendingService,
+            models: {
+                EmailBatch,
+                EmailRecipient,
+                Email,
+                Member
+            },
+            jobsService,
+            emailSegmenter,
+            emailRenderer,
+            db
+        });
+
+        this.service = new EmailService({
+            batchSendingService,
+            models: {
+                Email
+            },
+            settingsCache,
+            emailRenderer,
+            emailSegmenter,
+            limitService
+        });
+        
+        this.controller = new EmailController(this.service, {
+            models: {
+                Post,
+                Newsletter,
+                Email
+            }
+        });
+    }
+}
+
+module.exports = EmailServiceWrapper;

ghost/core/test/e2e-api/admin/emails.test.js

@@ -20,6 +20,7 @@ describe('Emails API', function () {
 
     beforeEach(function () {
         mockManager.mockEvents();
+        mockManager.mockLabsDisabled('emailStability');
     });
 
     afterEach(function () {

ghost/core/test/unit/server/services/email-service/EmailServiceWrapper.test.js

@@ -0,0 +1,12 @@
+describe('EmailServiceWrapper', function () {
+    it('Does not throw', async function () {
+        const offersService = require('../../../../../core/server/services/offers');
+        await offersService.init();
+
+        const memberService = require('../../../../../core/server/services/members');
+        await memberService.init();
+
+        const service = require('../../../../../core/server/services/email-service');
+        service.init();
+    });
+});

ghost/email-service/index.js

@@ -1,4 +1,8 @@
 module.exports = {
     EmailService: require('./lib/email-service'),
-    EmailController: require('./lib/email-controller')
+    EmailController: require('./lib/email-controller'),
+    EmailRenderer: require('./lib/email-renderer'),
+    EmailSegmenter: require('./lib/email-segmenter'),
+    SendingService: require('./lib/sending-service'),
+    BatchSendingService: require('./lib/batch-sending-service')
 };

ghost/email-service/lib/batch-sending-service.js

@@ -0,0 +1,366 @@
+const logging = require('@tryghost/logging');
+const ObjectID = require('bson-objectid').default;
+const errors = require('@tryghost/errors');
+const tpl = require('@tryghost/tpl');
+
+const messages = {
+    emailErrorPartialFailure: 'The email was only partially send',
+    emailError: 'Email failed to send'
+};
+
+/**
+ * @typedef {import('./sending-service')} SendingService
+ * @typedef {import('./email-segmenter')} EmailSegmenter
+ * @typedef {import('./email-renderer')} EmailRenderer
+ * @typedef {import('./email-renderer').MemberLike} MemberLike
+ * @typedef {object} JobsService
+ * @typedef {object} Email
+ * @typedef {object} Newsletter
+ * @typedef {object} Post
+ * @typedef {object} EmailBatch
+ */
+
+class BatchSendingService {
+    #emailRenderer;
+    #sendingService;
+    #emailSegmenter;
+    #jobsService;
+    #models;
+    #db;
+
+    /**
+     * @param {Object} dependencies 
+     * @param {EmailRenderer} dependencies.emailRenderer
+     * @param {SendingService} dependencies.sendingService
+     * @param {JobsService} dependencies.jobsService
+     * @param {EmailSegmenter} dependencies.emailSegmenter
+     * @param {object} dependencies.models
+     * @param {object} dependencies.models.EmailRecipient
+     * @param {EmailBatch} dependencies.models.EmailBatch
+     * @param {Email} dependencies.models.Email
+     * @param {object} dependencies.models.Member
+     * @param {object} dependencies.db
+     */
+    constructor({
+        emailRenderer,
+        sendingService,
+        jobsService,
+        emailSegmenter,
+        models,
+        db
+    }) {
+        this.#emailRenderer = emailRenderer;
+        this.#sendingService = sendingService;
+        this.#jobsService = jobsService;
+        this.#emailSegmenter = emailSegmenter;
+        this.#models = models;
+        this.#db = db;
+    }
+
+    /**
+     * Schedules a background job that sends the email in the background if it is pending or failed.
+     * @param {Email} email 
+     * @returns {void}
+     */
+    scheduleEmail(email) {
+        return this.#jobsService.addJob({
+            job: this.emailJob.bind(this),
+            data: {emailId: email.id},
+            offloaded: false
+        });
+    }
+
+    /**
+     * @private
+     * @param {{emailId: string}} data Data passed from the job service. We only need the emailId because we need to refetch the email anyway to make sure the status is right and 'locked'.
+     */
+    async emailJob({emailId}) {
+        logging.info(`Starting email job for email ${emailId}`);
+
+        // Check if email is 'pending' only + change status to submitting in one transaction.
+        // This allows us to have a lock around the email job that makes sure an email can only have one active job.
+        let email = await this.updateStatusLock(this.#models.Email, emailId, 'submitting', ['pending', 'failed']);
+        if (!email) {
+            logging.error(`Tried sending email that is not pending or failed ${emailId}`);
+            return;
+        }
+
+        try {
+            await this.sendEmail(email);
+
+            await email.save({
+                status: 'submitted',
+                submitted_at: new Date(),
+                error: null
+            }, {patch: true});
+        } catch (e) {
+            logging.error(`Error sending email ${email.id}: ${e.message}`);
+
+            // Edge case: Store error in email model (that are not caught by the batch)
+            await email.save({
+                status: 'failed',
+                error: e.message || 'Something went wrong while sending the email'
+            }, {patch: true});
+        }
+    }
+
+    /**
+     * @private
+     * @param {Email} email 
+     * @throws {errors.EmailError} If one of the batches fails
+     */
+    async sendEmail(email) {
+        logging.info(`Sending email ${email.id}`);
+
+        // Load required relations
+        const newsletter = await email.getLazyRelation('newsletter', {require: true});
+        const post = await email.getLazyRelation('post', {require: true});
+
+        let batches = await this.getBatches(email);
+        if (batches.length === 0) {
+            batches = await this.createBatches({email, newsletter, post});
+        }
+        await this.sendBatches({email, batches, post, newsletter});
+    }
+
+    /**
+     * @private
+     * @param {Email} email
+     * @returns {Promise<EmailBatch[]>}
+     */
+    async getBatches(email) {
+        logging.info(`Getting batches for email ${email.id}`);
+
+        return await this.#models.EmailBatch.findAll({filter: 'email_id:' + email.id});
+    }
+
+    /**
+     * @private
+     * @param {{email: Email, newsletter: Newsletter, post: Post}} data
+     * @returns {Promise<EmailBatch[]>}
+     */
+    async createBatches({email, post, newsletter}) {
+        logging.info(`Creating batches for email ${email.id}`);
+
+        const segments = await this.#emailRenderer.getSegments(post, newsletter);
+        const batches = [];
+        const BATCH_SIZE = 500;
+        let totalCount = 0;
+
+        for (const segment of segments) {
+            logging.info(`Creating batches for email ${email.id} segment ${segment}`);
+
+            const segmentFilter = this.#emailSegmenter.getMemberFilterForSegment(newsletter, email.get('recipient_filter'), segment);
+
+            // Avoiding Bookshelf for performance reasons
+            let members;
+            let lastId = null;
+
+            while (!members || lastId) {
+                logging.info(`Fetching members batch for email ${email.id} segment ${segment}, lastId: ${lastId}`);
+
+                const filter = segmentFilter + (lastId ? `+id:<${lastId}` : '');
+                members = await this.#models.Member.getFilteredCollectionQuery({filter, order: 'id DESC'}).select('members.id', 'members.uuid', 'members.email', 'members.name').limit(BATCH_SIZE + 1);
+
+                if (members.length > BATCH_SIZE) {
+                    lastId = members[members.length - 2].id;
+                } else {
+                    lastId = null;
+                }
+
+                if (members.length > 0) {
+                    totalCount += Math.min(members.length, BATCH_SIZE);
+                    const batch = await this.createBatch(email, segment, members.slice(0, BATCH_SIZE));
+                    batches.push(batch);
+                }
+            }
+        }
+
+        logging.info(`Created ${batches.length} batches for email ${email.id} with ${totalCount} recipients`);
+
+        if (email.get('email_count') !== totalCount) {
+            logging.error(`Email ${email.id} has wrong recipient count ${totalCount}, expected ${email.get('email_count')}. Updating the model.`);
+
+            // We update the email model because this will probably happen a few times because of the time difference
+            // between creating the email and sending it (or when the email failed initially and is retried a day later)
+            await email.save({
+                email_count: totalCount
+            }, {patch: true, require: false});
+        }
+        return batches;
+    }
+
+    /**
+     * @private
+     * @param {Email} email
+     * @param {import('./email-renderer').Segment} segment
+     * @param {object[]} members
+     * @returns {Promise<EmailBatch>}
+     */
+    async createBatch(email, segment, members, options) {
+        if (!options || !options.transacting) {
+            return this.#models.EmailBatch.transaction(async (transacting) => {
+                return this.createBatch(email, segment, members, {transacting});
+            });
+        }
+
+        logging.info(`Creating batch for email ${email.id} segment ${segment} with ${members.length} members`);
+
+        const batch = await this.#models.EmailBatch.add({
+            email_id: email.id,
+            member_segment: segment,
+            status: 'pending'
+        }, options);
+
+        const recipientData = [];
+
+        members.forEach((memberRow) => {
+            if (!memberRow.id || !memberRow.uuid || !memberRow.email) {
+                logging.warn(`Member row not included as email recipient due to missing data - id: ${memberRow.id}, uuid: ${memberRow.uuid}, email: ${memberRow.email}`);
+                return;
+            }
+
+            recipientData.push({
+                id: ObjectID().toHexString(),
+                email_id: email.id,
+                member_id: memberRow.id,
+                batch_id: batch.id,
+                member_uuid: memberRow.uuid,
+                member_email: memberRow.email,
+                member_name: memberRow.name
+            });
+        });
+
+        const insertQuery = this.#db.knex('email_recipients').insert(recipientData);
+
+        if (options.transacting) {
+            insertQuery.transacting(options.transacting);
+        }
+
+        logging.info(`Inserting ${recipientData.length} recipients for email ${email.id} batch ${batch.id}`);
+        await insertQuery;
+        return batch;
+    }
+
+    async sendBatches({email, batches, post, newsletter}) {
+        logging.info(`Sending ${batches.length} batches for email ${email.id}`);
+
+        // Loop batches and send them via the EmailProvider
+        // TODO: introduce concurrency when sending (need a replacement for bluebird)
+        let succeededCount = 0;
+        for (const batch of batches) {
+            if (await this.sendBatch({email, batch, post, newsletter})) {
+                succeededCount += 1;
+            }
+        }
+
+        if (succeededCount < batches.length) {
+            if (succeededCount > 0) {
+                throw new errors.EmailError({
+                    message: tpl(messages.emailErrorPartialFailure)
+                });
+            }
+            throw new errors.EmailError({
+                message: tpl(messages.emailError)
+            });
+        }
+    }
+
+    /**
+     * 
+     * @param {{email: Email, batch: EmailBatch, post: Post, newsletter: Newsletter}} data 
+     * @returns {Promise<boolean>} True when succeeded, false when failed with an error
+     */
+    async sendBatch({email, batch, post, newsletter}) {
+        logging.info(`Sending batch ${batch.id} for email ${email.id}`);
+
+        // Check the status of the email batch in a 'for update' transaction
+        batch = await this.updateStatusLock(this.#models.EmailBatch, batch.id, 'submitting', ['pending', 'failed']);
+        if (!batch) {
+            logging.error(`Tried sending email batch that is not pending or failed ${batch.id}; status: ${batch.get('status')}`);
+            return true;
+        }
+
+        let succeeded = false;
+
+        try {
+            const members = await this.getBatchMembers(batch.id);
+            const response = await this.#sendingService.send({
+                post,
+                newsletter,
+                segment: batch.get('member_segment'),
+                members
+            }, {
+                openTrackingEnabled: !!email.get('track_opens'),
+                clickTrackingEnabled: !!email.get('track_clicks')
+            });
+
+            await batch.save({
+                status: 'submitted',
+                provider_id: response.id
+            }, {patch: true, require: false});
+            succeeded = true;
+        } catch (err) {
+            logging.error(`Error sending email batch ${batch.id}`, err);
+
+            await batch.save({
+                status: 'failed'
+                // TODO: error should be instance of EmailProviderError (see IEmailProviderService) + we should read error message
+                // error_status_code: err.status_code,
+                // error_message: err.message_short,
+                // error_data: err.message_full
+            }, {patch: true, require: false});
+        }
+
+        // Mark as processed, even when failed
+        await this.#models.EmailRecipient
+            .where({batch_id: batch.id})
+            .save({processed_at: new Date()}, {patch: true, require: false});
+
+        return succeeded;
+    }
+
+    /**
+     * We don't want to pass EmailRecipient models to the sendingService.
+     * So we transform them into the MemberLike interface.
+     * That keeps the sending service nicely seperated so it isn't dependent on the batch sending data structure.
+     * @returns {Promise<MemberLike[]>}
+     */
+    async getBatchMembers(batchId) {
+        const models = await this.#models.EmailRecipient.findAll({filter: `batch_id:${batchId}`});
+        return models.map((model) => {
+            return {
+                id: model.get('member_id'),
+                uuid: model.get('member_uuid'),
+                email: model.get('member_email'),
+                name: model.get('member_name')
+            };
+        });
+    }
+
+    /**
+     * @private
+     * Update the status of an email or emailBatch to a given status, but first check if their current status is 'pending' or 'failed'.
+     * @param {object} Model Bookshelf model constructor
+     * @param {string} id id of the model
+     * @param {string} status set the status of the model to this value
+     * @param {string[]} allowedStatuses Check if the models current status is one of these values
+     * @returns {Promise<object|undefined>} The updated model. Undefined if the model didn't pass the status check.
+     */
+    async updateStatusLock(Model, id, status, allowedStatuses) {
+        let model;
+        await Model.transaction(async (transacting) => {
+            model = await Model.findOne({id}, {require: true, transacting, forUpdate: true});
+            if (!allowedStatuses.includes(model.get('status'))) {
+                model = undefined;
+                return;
+            }
+            await model.save({
+                status
+            }, {patch: true, transacting});
+        });
+        return model;
+    }
+}
+
+module.exports = BatchSendingService;

ghost/email-service/lib/email-controller.js

@@ -3,7 +3,8 @@ const tpl = require('@tryghost/tpl');
 
 const messages = {
     postNotFound: 'Post not found.',
-    noEmailsProvided: 'No emails provided.'
+    noEmailsProvided: 'No emails provided.',
+    emailNotFound: 'Email not found.'
 };
 
 class EmailController {
@@ -13,7 +14,7 @@ class EmailController {
     /**
      * 
      * @param {EmailService} service 
-     * @param {{models: {Post: any, Newsletter: any}}} dependencies
+     * @param {{models: {Post: any, Newsletter: any, Email: any}}} dependencies
      */
     constructor(service, {models}) {
         this.service = service;
@@ -60,6 +61,18 @@ class EmailController {
 
         return await this.service.sendTestEmail(post, newsletter, segment, emails);
     }
+
+    async retryFailedEmail(frame) {
+        const email = await this.models.Email.findOne(frame.data, {require: false});
+        
+        if (!email) {
+            throw new errors.NotFoundError({
+                message: tpl(messages.emailNotFound)
+            });
+        }
+
+        return await this.service.retryEmail(email);
+    }
 }
 
 module.exports = EmailController;

ghost/email-service/lib/email-renderer.js

@@ -0,0 +1,80 @@
+/* eslint-disable no-unused-vars */
+
+/**
+ * @typedef {string|null} Segment
+ * @typedef {object} Post
+ * @typedef {object} Newsletter
+ */
+
+/**
+ * @typedef {object} MemberLike
+ * @prop {string} id
+ * @prop {string} uuid
+ * @prop {string} email
+ * @prop {string} name
+ */
+
+/**
+ * @typedef {object} ReplacementDefinition
+ * @prop {string} token
+ * @prop {(member: MemberLike) => string} getValue
+ */
+
+/**
+ * @typedef {object} EmailRenderOptions
+ * @prop {boolean} clickTrackingEnabled
+ */
+
+/**
+ * @typedef {object} EmailBody
+ * @prop {string} html
+ * @prop {string} plaintext
+ * @prop {ReplacementDefinition[]} replacements
+ */
+
+class EmailRenderer {
+    /**
+		Not sure about this, but we need a method that can tell us which member segments are needed for a given post/email.
+        @param {Post} post
+        @param {Newsletter} newsletter
+        @returns {Promise<Segment[]>}
+	*/
+    async getSegments(post, newsletter) {
+        return [null];
+    }
+
+    /**
+     * 
+     * @param {Post} post 
+     * @param {Newsletter} newsletter 
+     * @param {Segment} segment 
+     * @param {EmailRenderOptions} options 
+     * @returns {Promise<EmailBody>}
+     */
+    async renderBody(post, newsletter, segment, options) {
+        return {
+            html: 'HTML',
+            plaintext: 'Plaintext',
+            replacements: []
+        };
+    }
+
+    getSubject(post, newsletter) {
+        return 'Subject';
+    }
+
+    getFromAddress(post, newsletter) {
+        return 'noreply@example.com'; // TODO
+    }
+
+    /**
+     * @param {Post} post
+     * @param {Newsletter} newsletter
+     * @returns {string|null}
+     */
+    getReplyToAddress(post, newsletter) {
+        return 'noreply@example.com'; // TODO
+    }
+}
+
+module.exports = EmailRenderer;

ghost/email-service/lib/email-segmenter.js

@@ -0,0 +1,74 @@
+const tpl = require('@tryghost/tpl');
+const errors = require('@tryghost/errors');
+
+const messages = {
+    noneFilterError: 'Cannot send email to "none" recipient filter',
+    newsletterVisibilityError: 'Unexpected visibility value "{value}". Use one of the valid: "members", "paid".'
+};
+
+/**
+ * @typedef {object} MembersRepository
+ * @prop {(options) => Promise<any>} list
+ */
+
+class EmailSegmenter {
+    #membersRepository;
+
+    /**
+     * 
+     * @param {object} dependencies 
+     * @param {MembersRepository} dependencies.membersRepository
+     */
+    constructor({
+        membersRepository
+    }) {
+        this.#membersRepository = membersRepository;
+    }
+
+    getMemberFilterForSegment(newsletter, emailRecipientFilter, segment) {
+        const filter = [`newsletters.id:${newsletter.id}`];
+    
+        switch (emailRecipientFilter) {
+        case 'all':
+            break;
+        case 'none':
+            throw new errors.InternalServerError({
+                message: tpl(messages.noneFilterError)
+            });
+        default:
+            filter.push(`(${emailRecipientFilter})`);
+            break;
+        }
+    
+        const visibility = newsletter.get('visibility');
+        switch (visibility) {
+        case 'members':
+            // No need to add a member status filter as the email is available to all members
+            break;
+        case 'paid':
+            filter.push(`status:-free`);
+            break;
+        default:
+            throw new errors.InternalServerError({
+                message: tpl(messages.newsletterVisibilityError, {
+                    value: visibility
+                })
+            });
+        }
+
+        if (segment) {
+            filter.push(`(${segment})`);
+        }
+    
+        return filter.join('+');
+    }
+    
+    async getMembersCount(newsletter, emailRecipientFilter, segment) {
+        const filter = this.getMemberFilterForSegment(newsletter, emailRecipientFilter, segment);
+        const {meta: {pagination: {total: membersCount}}} = await this.#membersRepository.list({filter});
+
+        return membersCount;
+    }
+}
+
+module.exports = EmailSegmenter;

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

@@ -1,17 +1,127 @@
 /* eslint-disable no-unused-vars */
 
+/**
+ * @typedef {object} Post
+ * @typedef {object} Email
+ * @typedef {object} LimitService
+ */
+
+const BatchSendingService = require('./batch-sending-service');
+const errors = require('@tryghost/errors');
+const tpl = require('@tryghost/tpl');
+const EmailRenderer = require('./email-renderer');
+const EmailSegmenter = require('./email-segmenter');
+
+const messages = {
+    archivedNewsletterError: 'Cannot send email to archived newsletters',
+    missingNewsletterError: 'The post does not have a newsletter relation'
+};
+
 class EmailService {
-    constructor(dependencies) {
-        // ...
+    #batchSendingService;
+    #models;
+    #settingsCache;
+    #emailRenderer;
+    #emailSegmenter;
+    #limitService;
+
+    /**
+     * 
+     * @param {object} dependencies 
+     * @param {BatchSendingService} dependencies.batchSendingService
+     * @param {object} dependencies.models
+     * @param {object} dependencies.models.Email
+     * @param {object} dependencies.settingsCache
+     * @param {EmailRenderer} dependencies.emailRenderer
+     * @param {EmailSegmenter} dependencies.emailSegmenter
+     * @param {LimitService} dependencies.limitService
+     */
+    constructor({
+        batchSendingService,
+        models,
+        settingsCache,
+        emailRenderer,
+        emailSegmenter,
+        limitService
+    }) {
+        this.#batchSendingService = batchSendingService;
+        this.#models = models;
+        this.#settingsCache = settingsCache;
+        this.#emailRenderer = emailRenderer;
+        this.#emailSegmenter = emailSegmenter;
+        this.#limitService = limitService;
     }
 
+    /**
+     * @private
+     */
+    async checkLimits() {
+        // Check host limit for allowed member count and throw error if over limit
+        // - do this even if it's a retry so that there's no way around the limit
+        if (this.#limitService.isLimited('members')) {
+            await this.#limitService.errorIfIsOverLimit('members');
+        }
+
+        // Check host limit for disabled emails or going over emails limit
+        if (this.#limitService.isLimited('emails')) {
+            await this.#limitService.errorIfWouldGoOverLimit('emails');
+        }
+    }
+
+    /**
+     * 
+     * @param {Post} post 
+     * @returns {Promise<Email>}
+     */
     async createEmail(post) {
-        // eslint-disable-next-line no-restricted-syntax
-        throw new Error('Not implemented');
+        let newsletter = await post.getLazyRelation('newsletter');
+        if (!newsletter) {
+            throw new errors.EmailError({
+                message: tpl(messages.missingNewsletterError)
+            });
+        }
+
+        if (newsletter.get('status') !== 'active') {
+            // A post might have been scheduled to an archived newsletter.
+            // Don't send it (people can't unsubscribe any longer).
+            throw new errors.EmailError({
+                message: tpl(messages.archivedNewsletterError)
+            });
+        }
+
+        const emailRecipientFilter = post.get('email_recipient_filter');
+
+        const email = await this.#models.Email.add({
+            post_id: post.id,
+            newsletter_id: newsletter.id,
+            status: 'pending',
+            submitted_at: new Date(),
+            track_opens: !!this.#settingsCache.get('email_track_opens'),
+            track_clicks: !!this.#settingsCache.get('email_track_clicks'),
+            feedback_enabled: !!newsletter.get('feedback_enabled'),
+            recipient_filter: emailRecipientFilter,
+            subject: this.#emailRenderer.getSubject(post, newsletter),
+            from: this.#emailRenderer.getFromAddress(post, newsletter),
+            replyTo: this.#emailRenderer.getReplyToAddress(post, newsletter),
+            email_count: await this.#emailSegmenter.getMembersCount(newsletter, emailRecipientFilter)
+        });
+
+        try {
+            await this.checkLimits();
+            this.#batchSendingService.scheduleEmail(email);
+        } catch (e) {
+            await email.save({
+                status: 'failed',
+                error: e.message || 'Something went wrong while scheduling the email'
+            }, {patch: true});
+        }
+
+        return email;
     }
     async retryEmail(email) {
-        // eslint-disable-next-line no-restricted-syntax
-        throw new Error('Not implemented');
+        await this.checkLimits();
+        this.#batchSendingService.scheduleEmail(email);
+        return email;
     }
 
     async previewEmail(post, newsletter, segment) {

ghost/email-service/lib/sending-service.js

@@ -0,0 +1,115 @@
+/**
+ * @typedef {object} EmailData
+ * @prop {string} html
+ * @prop {string} plaintext
+ * @prop {string} subject
+ * @prop {string} from
+ * @prop {string} [replyTo]
+ * @prop {Recipient[]} recipients
+ * 
+ * @typedef {object} IEmailProviderService
+ * @prop {(emailData: EmailData, options: EmailSendingOptions) => Promise<EmailProviderSuccessResponse>} send
+ * 
+ * @typedef {object} Post
+ * @typedef {object} Newsletter
+ */
+
+/**
+ * @typedef {import("./email-renderer")} EmailRenderer
+ */
+
+/**
+ * @typedef {object} EmailSendingOptions
+ * @prop {boolean} clickTrackingEnabled
+ * @prop {boolean} openTrackingEnabled
+ */
+
+/**
+ * @typedef {import("./email-renderer").MemberLike} MemberLike
+ */
+
+/**
+ * @typedef {object} Recipient
+ * @prop {string} email
+ * @prop {Replacement[]} replacements
+ */
+
+/**
+ * @typedef {object} Replacement
+ * @prop {string} token
+ * @prop {string} value
+ */
+
+/**
+ * @typedef {object} EmailProviderSuccessResponse
+ * @prop {string} id
+ */
+
+class SendingService {
+    #emailProvider;
+    #emailRenderer;
+
+    /**
+     * @param {object} dependencies
+     * @param {IEmailProviderService} dependencies.emailProvider
+     * @param {EmailRenderer} dependencies.emailRenderer
+     */
+    constructor({
+        emailProvider,
+        emailRenderer
+    }) {
+        this.#emailProvider = emailProvider;
+        this.#emailRenderer = emailRenderer;
+    }
+
+    /** 
+     * Send a given post, rendered for a given newsletter and segment to the members provided in the list
+     * @param {object} data
+     * @param {Post} data.post
+     * @param {Newsletter} data.newsletter
+     * @param {string|null} data.segment
+     * @param {MemberLike[]} data.members
+     * @param {EmailSendingOptions} options
+     * @returns {Promise<EmailProviderSuccessResponse>}
+    */
+    async send({post, newsletter, segment, members}, options) {
+        const emailBody = await this.#emailRenderer.renderBody(
+            post, 
+            newsletter, 
+            segment,
+            options
+        );
+
+        const recipients = this.buildRecipients(members, emailBody.replacements);
+        return await this.#emailProvider.send({
+            subject: this.#emailRenderer.getSubject(post, newsletter),
+            from: this.#emailRenderer.getFromAddress(post, newsletter),
+            replyTo: this.#emailRenderer.getReplyToAddress(post, newsletter) ?? undefined,
+            html: emailBody.html,
+            plaintext: emailBody.plaintext,
+            recipients
+        }, options);
+    }
+    
+    /**
+     * @private
+     * @param {MemberLike[]} members
+     * @param {import("./email-renderer").ReplacementDefinition[]} replacementDefinitions
+     * @returns {Recipient[]}
+     */
+    buildRecipients(members, replacementDefinitions) {
+        return members.map((member) => {
+            return {
+                email: member.email,
+                replacements: replacementDefinitions.map((def) => {
+                    return {
+                        token: def.token,
+                        value: def.getValue(member)
+                    };
+                })
+            };
+        });
+    }
+}
+
+module.exports = SendingService;

ghost/email-service/package.json

@@ -25,6 +25,7 @@
   },
   "dependencies": {
     "@tryghost/errors": "1.2.18",
-    "@tryghost/tpl": "0.1.19"
+    "@tryghost/tpl": "0.1.19",
+    "bson-objectid": "2.0.3"
   }
 }