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 {
	EventQueueEventId,
	EventQueueUpdate,
	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 { getAnnunciSchedaData } from "~/server/controllers/annunci.controller";
import { getRicevutaData } from "~/server/controllers/pagamenti.controller";
import {
	type MinimalSMSData,
	SendMessage,
} from "~/server/controllers/skebby.controller";
import { db } from "~/server/db";
import { NewMail, type NewMailProps } from "~/server/services/mailer";
import {
	convertHtmlToTypst,
	TypstGenerate,
} from "~/server/services/typst.service";
import { getKeydbClient } from "~/utils/keydb";
import {
	CONDIZIONI_CACHE_PREFIX,
	SCHEDA_ANNUNCIO_CACHE_PREFIX,
} from "../services/cache.service";

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

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

const generatePdfContent = async (params: GeneratedAttachmentTypes) => {
	switch (params.type) {
		case "ricevuta_cortesia": {
			const ricData = await getRicevutaData(params.ordineId);
			const pdf = await TypstGenerate({
				templateId: "receipt.typ",
				data: ricData.data,
			});
			return pdf.toString("base64");
		}

		case "condizioni": {
			const keybd = getKeydbClient();
			const cacheKey = `${CONDIZIONI_CACHE_PREFIX}${params.stringId}`;
			if (keybd) {
				const cached = await keybd.get(cacheKey);
				if (cached) {
					return cached;
				}
			}
			const data = await db
				.selectFrom("testi_e_stringhe")
				.selectAll()
				.where("stinga_id", "=", params.stringId)
				.executeTakeFirst();
			if (!data?.stringa_value) {
				throw new TRPCError({
					code: "NOT_FOUND",
					message: "Stringa non trovata",
				});
			}

			const stringa = convertHtmlToTypst(data.stringa_value);
			const pdf = await TypstGenerate({
				templateId: "testi.typ",
				data: { stringa },
			});
			const pdfBase64 = pdf.toString("base64");
			if (keybd) {
				await keybd.set(cacheKey, pdfBase64, "EX", 3600 * 24 * 5);
			}

			return pdfBase64;
		}
		case "scheda_annuncio": {
			const keybd = getKeydbClient();
			const cacheKey = `${SCHEDA_ANNUNCIO_CACHE_PREFIX}${params.annuncioId}`;
			if (keybd) {
				const cached = await keybd.get(cacheKey);
				if (cached) {
					return cached;
				}
			}
			const data = await getAnnunciSchedaData(params.annuncioId);
			const pdf = await TypstGenerate({
				templateId: "annuncio.typ",
				data,
			});
			const pdfBase64 = pdf.toString("base64");
			if (keybd) {
				await keybd.set(cacheKey, pdfBase64, "EX", 3600 * 5);
			}
			return pdfBase64;
		}
		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 getEmailQueue = async () => {
	try {
		const events = await db
			.selectFrom("event_queue")
			.selectAll("event_queue")
			.select((eb) => [
				eb
					.selectFrom("users")
					.select("username")
					.whereRef(
						"users.id",
						"=",
						sql`(event_queue.data->'data'->>'userId')::uuid`,
					)
					.as("username"),
			])
			.where(sql`data->>'tipo'`, "=", "email")
			.execute();
		return events;
	} catch (e) {
		throw new TRPCError({
			code: "INTERNAL_SERVER_ERROR",
			message: `Error while getting email events from queue, ${(e as Error).message}`,
		});
	}
};
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}`,
		});
	}
};

export const removeEventFromQueue = async (eventId: EventQueueEventId) => {
	try {
		await db
			.deleteFrom("event_queue")
			.where("event_id", "=", eventId)
			.execute();
		return true;
	} catch (e) {
		throw new TRPCError({
			code: "INTERNAL_SERVER_ERROR",
			message: `Error while removing event from queue, ${(e as Error).message}`,
			cause: e,
		});
	}
};
export const clearEmailQueue = async () => {
	try {
		await db
			.deleteFrom("event_queue")
			.where(sql`data->>'tipo'`, "=", "email")
			.execute();
		return true;
	} catch (e) {
		throw new TRPCError({
			code: "INTERNAL_SERVER_ERROR",
			message: `Error while clearing email events from queue, ${(e as Error).message}`,
			cause: e,
		});
	}
};
export const clearUserEmailQueue = async (userId: UsersId) => {
	try {
		await db
			.deleteFrom("event_queue")
			.where(sql`data->>'tipo'`, "=", "email")
			.where(sql`data->'data'->>'userId'`, "=", userId)
			.execute();
		return true;
	} catch (e) {
		throw new TRPCError({
			code: "INTERNAL_SERVER_ERROR",
			message: `Error while clearing user events from queue, ${(e as Error).message}`,
			cause: e,
		});
	}
};

export const updateEvent = async (
	eventId: EventQueueEventId,
	data: EventQueueUpdate,
) => {
	try {
		await db
			.updateTable("event_queue")
			.set(data)
			.where("event_id", "=", eventId)
			.execute();
		return true;
	} catch (e) {
		throw new TRPCError({
			code: "INTERNAL_SERVER_ERROR",
			message: `Error while updating event in queue, ${(e as Error).message}`,
			cause: e,
		});
	}
};