fix(notifications): prune dead push endpoints

1 files changed, 53 insertions, 16 deletions

diff --git a/src/trigger/notifications.ts b/src/trigger/notifications.ts
index 8a16624c..5b2453c2 100644
--- a/src/trigger/notifications.ts
+++ b/src/trigger/notifications.ts
@@ -2,28 +2,41 @@ import { createClient } from "@supabase/supabase-js";
 import { envvars, schedules } from "@trigger.dev/sdk";
 import * as webpush from "web-push";
 
+const isExpiredSubscriptionError = (error: unknown) => {
+	const statusCode =
+		typeof error === "object" &&
+		error !== null &&
+		"statusCode" in error &&
+		typeof error.statusCode === "number"
+			? error.statusCode
+			: null;
+
+	return statusCode === 404 || statusCode === 410;
+};
+
 export const notificationsTask = schedules.task({
 	id: "notifications",
 	run: async (_payload, { ctx }) => {
 		const environment = ctx.environment.slug;
 		const triggerProjectReference = ctx.project.ref;
+		const supabase = createClient(
+			(
+				await envvars.retrieve(
+					triggerProjectReference,
+					environment,
+					"SUPABASE_URL",
+				)
+			).value,
+			(
+				await envvars.retrieve(
+					triggerProjectReference,
+					environment,
+					"SUPABASE_SERVICE_ROLE_KEY",
+				)
+			).value,
+		);
 		const getUserSubscriptions = async () => {
-			const { data, error } = await createClient(
-				(
-					await envvars.retrieve(
-						triggerProjectReference,
-						environment,
-						"SUPABASE_URL",
-					)
-				).value,
-				(
-					await envvars.retrieve(
-						triggerProjectReference,
-						environment,
-						"SUPABASE_SERVICE_ROLE_KEY",
-					)
-				).value,
-			)
+			const { data, error } = await supabase
 				.from("user_notifications")
 				.select("*");
 
@@ -31,6 +44,18 @@ export const notificationsTask = schedules.task({
 
 			return data;
 		};
+		const deleteUserSubscription = async (
+			userId: number,
+			fingerprint: string,
+		) => {
+			await supabase
+				.from("user_notifications")
+				.delete()
+				.eq("user_id", userId)
+				.eq("fingerprint", fingerprint);
+		};
+
+		const transientErrors: unknown[] = [];
 
 		webpush.setVapidDetails(
 			(
@@ -60,9 +85,21 @@ export const notificationsTask = schedules.task({
 			try {
 				await webpush.sendNotification(subscription.subscription, ".");
 			} catch (error) {
+				if (isExpiredSubscriptionError(error))
+					await deleteUserSubscription(
+						subscription.user_id,
+						subscription.fingerprint,
+					);
+				else transientErrors.push(error);
+
 				console.error(error);
 			}
 
+		if (transientErrors.length > 0)
+			throw new Error("Transient push delivery failures occurred", {
+				cause: transientErrors[0],
+			});
+
 		return {};
 	},
 });