javascript - E-postkøen fortsetter å sende ut duplikate e-poster

Rediger

Så jeg fant det ut, jeg overså følgende linje

                const emails = [...this.emailQueue];

                for (let i = 0; i < emails.length; i++) {
                        const [emailUuid, { attempts, emailLogEvent, ...emailConfig }] = emails[i];
 
+                       processEmails[emailUuid] = false;
                        emailClient(emailConfig)
                                .then((result) => {

Jeg prøver å implementere en e-postkø, men det fungerer ikke helt som forventet. Ideen er å svare fra et endepunkt så raskt som mulig og deretter bare la køadministratoren forsøke å behandle køen i bakgrunnen, i dette tilfellet sende ut en e-post.

Et av problemene er at intervallet fortsetter å kjøre når this.running er satt til true. Et annet problem er at en e-post noen ganger sendes ut flere ganger, når den bare skulle sendes ut én gang.

Jeg har også prøvd å bruke Redis for dette fordi jeg kjenner igjen lukkeproblemet med initialize og også så Bull som en potensiell løsning, men jeg prøver å unngå så mange unødvendige biblioteker som mulig. Ideelt sett vil jeg lage en slags baseklasse som jeg kan gjenbruke for ulike køer, for eksempel en SMS-kø, ikke bare en e-postkø. Men foreløpig antar jeg at jeg bare prøver å få denne e-postkøen til å fungere.

Hva gjør jeg feil her? Her er koden min:

import emailClient from 'services/emailClient';
import randomUuid from 'services/randomUuid'

export interface EmailClientConfig {
    to: string;
}

const MAX_ATTEMPTS_FOR_EMAILS = 5;
class EmailQueueManager {
    private running = false;
    private emailQueue = new Map();

    private handleRemoveSentEmailsFromQueue(sentEmailUuids: string[]) {
        // delete each sent out email from queue
        sentEmailUuids.forEach((sentEmailUuid) => {
            this.emailQueue.delete(sentEmailUuid);
        });

        // set state to "ready to process"
        this.running = false;
    }

    private async handleProcessEmailQueue() {
        // determines if queue is currently being executed
        this.running = true;
        const processedEmails: Record<string, boolean> = {};
        // list of successfully sent emails
        const sentEmailUuids: string[] = [];

        // do nothing if currently being executed, so as
        // to not send out duplicate emails
        if (!this.emailQueue.size) {
            this.running = false;
            return;
        }

        const emails = [...this.emailQueue];

        for (let i = 0; i < emails.length; i++) {
            const [emailUuid, { attempts, emailLogEvent, ...emailConfig }] = emails[i];

            emailClient(emailConfig)
                .then((result) => {
                    sentEmailUuids.push(emailUuid);

                    // maybe do something else
                })
                .catch((err) => {
                    if (attempts > MAX_ATTEMPTS_FOR_EMAILS) {
                        // remove from queue permanently if
                        // attempt count exceeds what is allowed
                        this.emailQueue.delete(emailUuid);
                    } else {
                        // send back to queue and attempt to send email out again
                        this.emailQueue.set(emailUuid, {
                            attempts: attempts + 1,
                            emailLogEvent,
                            ...emailConfig,
                        });
                    }
                })
                .finally(() => {
                    // flag email as "processed", regardless of whether
                    // or not is was successfully sent
                    processedEmails[emailUuid] = true;
                });
        }

        // check every 250ms if all emails in queue
        // were processed (not necessarily successfully sent out)
        const checkProcessedEmailsInterval = setInterval(() => {
            if (Object.values(processedEmails).every(Boolean)) {
                this.handleRemoveSentEmailsFromQueue(sentEmailUuids);
                clearInterval(checkProcessedEmailsInterval);
            }
        }, 250);
    }
    public addEmailToQueue({
        emailConfig,
        emailLogEvent,
    }: {
        emailConfig: EmailClientConfig;
        emailLogEvent: string;
    }) {
        // add email to queue
        this.emailQueue.set(randomUuid(), { attempts: 0, emailLogEvent, ...emailConfig });
        // and immediately process the queue
        this.handleProcessEmailQueue();
    }

    public initialize() {
        // check every minute if there are
        // emails in queue that need to be processed
        setInterval(() => {
            if (!this.running) {
                this.handleProcessEmailQueue();
            }
        }, 1000 * 60);
    }
}

const emailQueueManager = new EmailQueueManager();

// instantiate the interval
emailQueueManager.initialize();

export default emailQueueManager;

Og jeg bruker dette slik:

import emailQueueManager from 'src/managers/queues/emailQueue';

. . .
async function someEndpoint(req, res) {
    // do stuff
    emailQueueManager.addEmailToQueue({
        emailConfig: {
            to: '[email ',
        },
        emailLogEvent: 'sent_email',
    });
    // do more stuff

    res.end()
}
. . .