Blame for apps/api/dist/queue/workers.js

"use strict";
var __createBinding = (this && this.__createBinding) || (Object.create ? (function(o, m, k, k2) {
    if (k2 === undefined) k2 = k;
    var desc = Object.getOwnPropertyDescriptor(m, k);
    if (!desc || ("get" in desc ? !m.__esModule : desc.writable || desc.configurable)) {
      desc = { enumerable: true, get: function() { return m[k]; } };
    }
    Object.defineProperty(o, k2, desc);
}) : (function(o, m, k, k2) {
    if (k2 === undefined) k2 = k;
    o[k2] = m[k];
}));
var __setModuleDefault = (this && this.__setModuleDefault) || (Object.create ? (function(o, v) {
    Object.defineProperty(o, "default", { enumerable: true, value: v });
}) : function(o, v) {
    o["default"] = v;
});
var __decorate = (this && this.__decorate) || function (decorators, target, key, desc) {
    var c = arguments.length, r = c < 3 ? target : desc === null ? desc = Object.getOwnPropertyDescriptor(target, key) : desc, d;
    if (typeof Reflect === "object" && typeof Reflect.decorate === "function") r = Reflect.decorate(decorators, target, key, desc);
    else for (var i = decorators.length - 1; i >= 0; i--) if (d = decorators[i]) r = (c < 3 ? d(r) : c > 3 ? d(target, key, r) : d(target, key)) || r;
    return c > 3 && r && Object.defineProperty(target, key, r), r;
};
var __importStar = (this && this.__importStar) || (function () {
    var ownKeys = function(o) {
        ownKeys = Object.getOwnPropertyNames || function (o) {
            var ar = [];
            for (var k in o) if (Object.prototype.hasOwnProperty.call(o, k)) ar[ar.length] = k;
            return ar;
        };
        return ownKeys(o);
    };
    return function (mod) {
        if (mod && mod.__esModule) return mod;
        var result = {};
        if (mod != null) for (var k = ownKeys(mod), i = 0; i < k.length; i++) if (k[i] !== "default") __createBinding(result, mod, k[i]);
        __setModuleDefault(result, mod);
        return result;
    };
})();
var __metadata = (this && this.__metadata) || function (k, v) {
    if (typeof Reflect === "object" && typeof Reflect.metadata === "function") return Reflect.metadata(k, v);
};
var __param = (this && this.__param) || function (paramIndex, decorator) {
    return function (target, key) { decorator(target, key, paramIndex); }
};
var EmailWorker_1, ReminderWorker_1, AssetProcessingWorker_1, OutreachAutomationWorker_1;
Object.defineProperty(exports, "__esModule", { value: true });
exports.OutreachAutomationWorker = exports.AssetProcessingWorker = exports.ReminderWorker = exports.EmailWorker = exports.QUEUE_NAMES = void 0;
const bullmq_1 = require("@nestjs/bullmq");
const bullmq_2 = require("@nestjs/bullmq");
const common_1 = require("@nestjs/common");
const bullmq_3 = require("bullmq");
const path = __importStar(require("path"));
const fs = __importStar(require("fs/promises"));
const os = __importStar(require("os"));
const crypto_1 = require("crypto");
const prisma_service_1 = require("../prisma/prisma.service");
const escape_html_1 = require("../common/escape-html");
const email_service_1 = require("../email/email.service");
exports.QUEUE_NAMES = {
    EMAIL: "email",
    REMINDER: "reminder",
    ASSET_PROCESSING: "asset-processing",
    OUTREACH_AUTOMATION: "outreach-automation",
    VERIFICATION: "verification",
};
let EmailWorker = EmailWorker_1 = class EmailWorker extends bullmq_1.WorkerHost {
    emailService;
    logger = new common_1.Logger(EmailWorker_1.name);
    constructor(emailService) {
        super();
        this.emailService = emailService;
    }
    async process(job) {
        this.logger.log(`Processing email job ${job.id}: ${job.data.subject} → ${job.data.to}`);
        try {
            const result = await this.emailService.sendEmail({
                to: job.data.to,
                subject: job.data.subject,
                bodyHtml: job.data.bodyHtml,
                bodyText: job.data.bodyText,
            });
            if (result.success) {
                this.logger.log(`Email sent successfully to ${job.data.to}`);
                return result;
            }
            throw new Error("Email delivery returned failure");
        }
        catch (error) {
            this.logger.error(`Failed to send email to ${job.data.to}: ${error instanceof Error ? error.message : String(error)}`);
            throw new Error(`Email delivery failed: ${error instanceof Error ? error.message : String(error)}`);
        }
    }
};
exports.EmailWorker = EmailWorker;
exports.EmailWorker = EmailWorker = EmailWorker_1 = __decorate([
    (0, common_1.Injectable)(),
    (0, bullmq_1.Processor)(exports.QUEUE_NAMES.EMAIL),
    __metadata("design:paramtypes", [email_service_1.EmailService])
], EmailWorker);
let ReminderWorker = ReminderWorker_1 = class ReminderWorker extends bullmq_1.WorkerHost {
    prisma;
    emailQueue;
    logger = new common_1.Logger(ReminderWorker_1.name);
    constructor(prisma, emailQueue) {
        super();
        this.prisma = prisma;
        this.emailQueue = emailQueue;
    }
    async process(job) {
        this.logger.log(`Processing ${job.data.type} reminder for ${job.data.recipientName} (${job.data.recipientEmail})`);
        const { subject, bodyHtml, bodyText } = this.renderReminder(job.data);
        if (this.emailQueue) {
            await this.emailQueue.add("send-reminder", {
                to: job.data.recipientEmail,
                subject,
                bodyHtml,
                bodyText,
            }, {
                attempts: 3,
                backoff: { type: "exponential", delay: 60_000 },
                removeOnComplete: { age: 86_400, count: 100 },
                removeOnFail: { age: 7 * 86_400 },
            });
        }
        else {
            this.logger.warn("Email queue not registered; reminder email will not be sent.");
        }
        return { success: true };
    }
    renderReminder(data) {
        const safeName = (0, escape_html_1.escapeHtml)(data.recipientName || "there");
        const greeting = `Hi ${safeName},`;
        switch (data.type) {
            case "booking": {
                const startTime = typeof data.details.startTime === "string"
                    ? (0, escape_html_1.escapeHtml)(data.details.startTime)
                    : "your scheduled time";
                const subject = `Reminder: Your recording is coming up at ${startTime}`;
                const bodyText = `${greeting}\n\n` +
                    `This is a friendly reminder of your upcoming recording session at ${startTime}.\n` +
                    `Please join a few minutes early to test your audio setup.\n\n` +
                    `— MicProtocol`;
                return {
                    subject,
                    bodyText,
                    bodyHtml: `<p>${(0, escape_html_1.escapeHtml)(bodyText).replace(/\n/g, "<br/>")}</p>`,
                };
            }
            case "asset-deadline": {
                const due = typeof data.details.dueDate === "string"
                    ? (0, escape_html_1.escapeHtml)(data.details.dueDate)
                    : "soon";
                const subject = `Action required: assets due by ${due}`;
                const bodyText = `${greeting}\n\n` +
                    `We still need some assets from you before ${due}. ` +
                    `Please log in to your guest portal to upload them.\n\n` +
                    `— MicProtocol`;
                return {
                    subject,
                    bodyText,
                    bodyHtml: `<p>${(0, escape_html_1.escapeHtml)(bodyText).replace(/\n/g, "<br/>")}</p>`,
                };
            }
            case "followup": {
                const subject = `Quick follow-up`;
                const bodyText = `${greeting}\n\n` +
                    `Just checking in to see if you had a chance to review our previous message. ` +
                    `Happy to answer any questions you might have.\n\n` +
                    `— MicProtocol`;
                return {
                    subject,
                    bodyText,
                    bodyHtml: `<p>${(0, escape_html_1.escapeHtml)(bodyText).replace(/\n/g, "<br/>")}</p>`,
                };
            }
        }
    }
};
exports.ReminderWorker = ReminderWorker;
exports.ReminderWorker = ReminderWorker = ReminderWorker_1 = __decorate([
    (0, common_1.Injectable)(),
    (0, bullmq_1.Processor)(exports.QUEUE_NAMES.REMINDER),
    __param(1, (0, common_1.Optional)()),
    __param(1, (0, bullmq_2.InjectQueue)(exports.QUEUE_NAMES.EMAIL)),
    __metadata("design:paramtypes", [prisma_service_1.PrismaService,
        bullmq_3.Queue])
], ReminderWorker);
let AssetProcessingWorker = AssetProcessingWorker_1 = class AssetProcessingWorker extends bullmq_1.WorkerHost {
    prisma;
    logger = new common_1.Logger(AssetProcessingWorker_1.name);
    sharpModule = null;
    ffmpegModule = null;
    constructor(prisma) {
        super();
        this.prisma = prisma;
    }
    async loadSharp() {
        if (this.sharpModule)
            return this.sharpModule;
        try {
            const mod = await Promise.resolve().then(() => __importStar(require("sharp")));
            this.sharpModule = mod.default;
            return this.sharpModule;
        }
        catch (err) {
            this.logger.warn("Sharp is not installed; image processing will be skipped.");
            return null;
        }
    }
    async loadFfmpeg() {
        if (this.ffmpegModule)
            return this.ffmpegModule;
        try {
            this.ffmpegModule = (await Promise.resolve().then(() => __importStar(require("fluent-ffmpeg"))));
            return this.ffmpegModule;
        }
        catch (err) {
            this.logger.warn("fluent-ffmpeg is not installed; audio/video transcoding will be skipped.");
            return null;
        }
    }
    async process(job) {
        this.logger.log(`Processing asset ${job.data.submissionId}: ${job.data.operations.join(", ")}`);
        const processedUrls = [];
        const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "mp-asset-"));
        let downloadedPath = null;
        try {
            if (job.data.fileUrl && /^https?:\/\//.test(job.data.fileUrl)) {
                downloadedPath = await this.downloadToTmp(job.data.fileUrl, tmpDir);
            }
            for (const operation of job.data.operations) {
                const outPath = path.join(tmpDir, `${operation}-${(0, crypto_1.randomUUID)()}.${this.outputExt(operation)}`);
                try {
                    switch (operation) {
                        case "resize":
                            await this.resize(downloadedPath ?? Buffer.from(""), outPath, job.data.options?.width, job.data.options?.height);
                            break;
                        case "background-removal":
                            await this.flattenBackground(downloadedPath, outPath);
                            break;
                        case "transcode":
                            await this.transcodeAudio(downloadedPath, outPath, job.data.options?.bitrateKbps ?? 128);
                            break;
                        case "validate":
                            await this.validate(downloadedPath, job.data.fileType);
                            continue;
                    }
                    processedUrls.push(`file://${outPath}`);
                }
                catch (opErr) {
                    this.logger.error(`Asset operation "${operation}" failed for submission ${job.data.submissionId}: ${opErr instanceof Error ? opErr.message : String(opErr)}`);
                }
            }
            if (processedUrls.length > 0) {
                await this.prisma.assetSubmission
                    .update({
                    where: { id: job.data.submissionId },
                    data: {
                        fileUrl: processedUrls[0].replace("file://", ""),
                    },
                })
                    .catch((err) => {
                    this.logger.warn(`Could not update submission ${job.data.submissionId}: ${err instanceof Error ? err.message : String(err)}`);
                });
            }
            return { success: true, processedUrls };
        }
        finally {
            fs.rm(tmpDir, { recursive: true, force: true }).catch(() => undefined);
        }
    }
    async downloadToTmp(url, tmpDir) {
        const response = await fetch(url);
        if (!response.ok) {
            throw new Error(`Failed to download asset (${response.status} ${response.statusText})`);
        }
        const arrayBuffer = await response.arrayBuffer();
        const buffer = Buffer.from(arrayBuffer);
        const filename = `source-${(0, crypto_1.randomUUID)()}.bin`;
        const outPath = path.join(tmpDir, filename);
        await fs.writeFile(outPath, buffer);
        return outPath;
    }
    outputExt(operation) {
        switch (operation) {
            case "resize":
            case "background-removal":
                return "png";
            case "transcode":
                return "mp3";
            default:
                return "bin";
        }
    }
    async resize(input, outPath, width = 1024, height = 1024) {
        const sharp = await this.loadSharp();
        if (!sharp) {
            throw new Error("sharp is not available");
        }
        await sharp(input)
            .resize(width, height, { fit: "inside", withoutEnlargement: true })
            .png({ quality: 90 })
            .toFile(outPath);
    }
    async flattenBackground(input, outPath) {
        if (!input) {
            throw new Error("No input file for background flattening");
        }
        const sharp = await this.loadSharp();
        if (!sharp) {
            throw new Error("sharp is not available");
        }
        await sharp(input)
            .flatten({ background: { r: 255, g: 255, b: 255 } })
            .png()
            .toFile(outPath);
    }
    async transcodeAudio(input, outPath, bitrateKbps) {
        if (!input) {
            throw new Error("No input file for transcoding");
        }
        const ffmpeg = await this.loadFfmpeg();
        if (!ffmpeg) {
            throw new Error("fluent-ffmpeg is not available");
        }
        await new Promise((resolve, reject) => {
            ffmpeg(input)
                .audioBitrate(`${bitrateKbps}k`)
                .format("mp3")
                .save(outPath)
                .on("end", () => resolve())
                .on("error", (err) => reject(err));
        });
    }
    async validate(input, fileType) {
        if (!input) {
            throw new Error("No input file to validate");
        }
        const isImage = fileType.startsWith("image/");
        const isAudio = fileType.startsWith("audio/");
        const isVideo = fileType.startsWith("video/");
        if (isImage) {
            const sharp = await this.loadSharp();
            if (!sharp) {
                throw new Error("sharp is not available for image validation");
            }
            const meta = await sharp(input).metadata();
            if (!meta.width || !meta.height) {
                throw new Error("Image metadata missing dimensions");
            }
            if (meta.width < 200 || meta.height < 200) {
                throw new Error(`Image too small: ${meta.width}x${meta.height} (minimum 200x200)`);
            }
        }
        else if (isAudio || isVideo) {
            const ffmpeg = await this.loadFfmpeg();
            if (!ffmpeg) {
                throw new Error("fluent-ffmpeg is not available for media validation");
            }
            await new Promise((resolve, reject) => {
                ffmpeg.ffprobe(input, (err, data) => {
                    if (err) {
                        reject(err);
                        return;
                    }
                    const stream = data.streams?.[0];
                    if (!stream) {
                        reject(new Error("No media stream found in file"));
                        return;
                    }
                    resolve();
                });
            });
        }
        else {
            const stat = await fs.stat(input);
            if (stat.size === 0) {
                throw new Error("Asset file is empty");
            }
        }
    }
};
exports.AssetProcessingWorker = AssetProcessingWorker;
exports.AssetProcessingWorker = AssetProcessingWorker = AssetProcessingWorker_1 = __decorate([
    (0, common_1.Injectable)(),
    (0, bullmq_1.Processor)(exports.QUEUE_NAMES.ASSET_PROCESSING),
    __metadata("design:paramtypes", [prisma_service_1.PrismaService])
], AssetProcessingWorker);
let OutreachAutomationWorker = OutreachAutomationWorker_1 = class OutreachAutomationWorker extends bullmq_1.WorkerHost {
    prisma;
    emailQueue;
    logger = new common_1.Logger(OutreachAutomationWorker_1.name);
    constructor(prisma, emailQueue) {
        super();
        this.prisma = prisma;
        this.emailQueue = emailQueue;
    }
    async process(job) {
        this.logger.log(`Processing outreach automation: ${job.data.type} for sequence ${job.data.sequenceId} step ${job.data.step}`);
        const sequence = await this.prisma.outreachSequence.findUnique({
            where: { id: job.data.sequenceId },
            include: {
                guest: { select: { id: true, fullName: true, email: true } },
                messages: { orderBy: { createdAt: "asc" } },
            },
        });
        if (!sequence) {
            throw new Error(`Outreach sequence ${job.data.sequenceId} not found`);
        }
        if (sequence.status !== "active") {
            this.logger.log(`Sequence ${sequence.id} is no longer active (status=${sequence.status}); skipping.`);
            return { success: true };
        }
        switch (job.data.type) {
            case "send-initial": {
                const next = sequence.messages.find((m) => !m.sentAt);
                if (next) {
                    await this.enqueueEmail(next.id, sequence.guest.email, {
                        subject: next.subject,
                        bodyHtml: next.bodyHtml,
                        bodyText: next.bodyText,
                    });
                }
                break;
            }
            case "send-followup": {
                const target = sequence.messages[job.data.step - 1];
                if (target && !target.sentAt) {
                    await this.enqueueEmail(target.id, sequence.guest.email, {
                        subject: target.subject,
                        bodyHtml: target.bodyHtml,
                        bodyText: target.bodyText,
                    });
                }
                break;
            }
            case "check-response": {
                this.logger.debug(`Response check for sequence ${sequence.id} (guest ${sequence.guest.email})`);
                break;
            }
        }
        return { success: true };
    }
    async enqueueEmail(messageId, to, payload) {
        if (!this.emailQueue) {
            this.logger.warn("Email queue not registered; outreach email will not be sent.");
            return;
        }
        await this.emailQueue.add("outreach-send", {
            to,
            subject: payload.subject,
            bodyHtml: payload.bodyHtml,
            bodyText: payload.bodyText,
            messageId,
        }, {
            attempts: 3,
            backoff: { type: "exponential", delay: 60_000 },
            removeOnComplete: { age: 86_400, count: 200 },
            removeOnFail: { age: 7 * 86_400 },
        });
    }
};
exports.OutreachAutomationWorker = OutreachAutomationWorker;
exports.OutreachAutomationWorker = OutreachAutomationWorker = OutreachAutomationWorker_1 = __decorate([
    (0, common_1.Injectable)(),
    (0, bullmq_1.Processor)(exports.QUEUE_NAMES.OUTREACH_AUTOMATION),
    __param(1, (0, common_1.Optional)()),
    __param(1, (0, bullmq_2.InjectQueue)(exports.QUEUE_NAMES.EMAIL)),
    __metadata("design:paramtypes", [prisma_service_1.PrismaService,
        bullmq_3.Queue])
], OutreachAutomationWorker);
//# sourceMappingURL=workers.js.map