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 type { NewEventQueue } from "~/schemas/public/EventQueue";
import {
	type MinimalSMSData,
	SendMessage,
} from "~/server/controllers/skebby.controller";
import { db } from "~/server/db";
import { NewMail, type NewMailProps } from "~/server/services/mailer";

type Email_EventData = {
	tipo: "email";
	data: NewMailProps;
};

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;
};

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

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

		switch (data.tipo) {
			case "email":
				await NewMail({
					...data.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();
	}
};