infoalloggi-monorepo/apps/infoalloggi/src/server/controllers/event_queue.controller.ts

import type { Override } from "TypeHelpers.type";
import { TRPCError } from "@trpc/server";
import { add, set } from "date-fns";
import { sql } from "kysely";
import type { Attachment } from "nodemailer/lib/mailer";
import type { AnnunciId } from "~/schemas/public/Annunci";
import type { NewEventQueue } from "~/schemas/public/EventQueue";
import type { OrdiniOrdineId } from "~/schemas/public/Ordini";
import type { TestiEStringheStingaId } from "~/schemas/public/TestiEStringhe";
import type { UsersId } from "~/schemas/public/Users";
import {
	type MinimalSMSData,
	SendMessage,
} from "~/server/controllers/skebby.controller";
import { db } from "~/server/db";
import { NewMail, type NewMailProps } from "~/server/services/mailer";
import {
	genPdfCondizioniBase64,
	genPdfRicevutaCortesiaBase64,
	genPdfSchedaAnnuncioBase64,
} from "../services/puppeteer.service";

type GeneratedAttachmentTypes =
	| { type: "scheda_annuncio"; annuncioId: AnnunciId }
	| { type: "ricevuta_cortesia"; ordineId: OrdiniOrdineId }
	| { type: "condizioni"; stringId: TestiEStringheStingaId };

type CustomAttachment = Attachment & { generate?: GeneratedAttachmentTypes };
type Email_EventData = {
	tipo: "email";
	data: NewMailProps & {
		mail: {
			attachments?: CustomAttachment[];
		};
	};
};

const generatePdfContent = async (params: GeneratedAttachmentTypes) => {
	switch (params.type) {
		case "ricevuta_cortesia":
			return await genPdfRicevutaCortesiaBase64(params.ordineId);
		case "condizioni":
			return await genPdfCondizioniBase64(params.stringId);
		case "scheda_annuncio":
			return await genPdfSchedaAnnuncioBase64(params.annuncioId);
		default:
			throw new Error(`Unknown PDF type: `);
	}
};

type SMS_EventData = {
	tipo: "sms";
	data: MinimalSMSData;
};

export const canSendNotification = () => {
	const d = new Date();
	const hour = d.getHours();
	return hour >= 9 && hour <= 20;
};

export const nextTimeForNotification = () => {
	const d = new Date();
	const targetDay = d.getHours() < 9 ? d : add(d, { days: 1 });
	const lock_expires = set(targetDay, {
		hours: 9,
		minutes: 0,
	});
	return lock_expires;
};

type EventDataTypes = Email_EventData | SMS_EventData;

type NewEventQueueProps = Override<NewEventQueue, { data: EventDataTypes }>;

export const addEventToQueue = async (data: NewEventQueueProps) => {
	try {
		await db.insertInto("event_queue").values(data).execute();
	} catch (e) {
		throw new TRPCError({
			code: "INTERNAL_SERVER_ERROR",
			message: `Error while adding event to queue, ${(e as Error).message}`,
		});
	}
};

const getEventsFromQueue = async () => {
	try {
		const events = await db
			.selectFrom("event_queue")
			.selectAll()
			.where("status", "=", "pending")
			.where("event_queue.lock_expires", "<=", new Date())
			.execute();
		return events;
	} catch (e) {
		throw new TRPCError({
			code: "INTERNAL_SERVER_ERROR",
			message: `Error while getting events from queue, ${(e as Error).message}`,
		});
	}
};

const processEmailEvent = async (event: Email_EventData) => {
	const { template, mail, userId } = event.data;

	// Generate attachments on-demand
	if (mail.attachments && mail.attachments.length > 0) {
		mail.attachments = await Promise.all(
			mail.attachments.map(async (attachment: CustomAttachment) => {
				// If attachment has generate metadata, generate it now
				if (attachment.generate) {
					const content = await generatePdfContent(attachment.generate);
					return {
						filename: attachment.filename,
						content,
						encoding: attachment.encoding,
						contentType: attachment.contentType,
					};
				}
				// If already has content, return as-is
				return attachment;
			}),
		);
	}

	await NewMail({
		template,
		mail,
		userId,
	});
};

export const processEvents = async () => {
	const events = await getEventsFromQueue();
	if (events.length === 0) {
		return { message: "No events to process" };
	}
	for (const e of events) {
		const data = e.data as EventDataTypes;

		switch (data.tipo) {
			case "email": {
				await processEmailEvent(data);
				break;
			}

			case "sms":
				await SendMessage(data.data);
				break;
		}
		// Mark event as processed
		await db
			.deleteFrom("event_queue")
			.where("event_id", "=", e.event_id)
			.execute();
	}
	return { message: `${events.length} events processed` };
};

export const userEmailQueue = async (userId: UsersId) => {
	try {
		const events = await db
			.selectFrom("event_queue")
			.selectAll()
			.where(sql`data->>'tipo'`, "=", "email")
			.where(sql`data->'data'->>'userId'`, "=", userId)
			.execute();
		return events;
	} catch (e) {
		throw new TRPCError({
			code: "INTERNAL_SERVER_ERROR",
			message: `Error while getting user events from queue, ${(e as Error).message}`,
		});
	}
};