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 || !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; 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}`, }); } }; 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 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, }); } };