1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
|
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;
};
const subscriptionEndpoint = (subscription: unknown) =>
typeof subscription === "object" &&
subscription !== null &&
"endpoint" in subscription &&
typeof subscription.endpoint === "string" &&
subscription.endpoint.length
? subscription.endpoint
: null;
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 supabase
.from("user_notifications")
.select("*");
if (error) return [];
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[] = [];
const seenEndpoints = new Set<string>();
webpush.setVapidDetails(
(
await envvars.retrieve(
triggerProjectReference,
environment,
"VAPID_SUBJECT",
)
).value,
(
await envvars.retrieve(
triggerProjectReference,
environment,
"VAPID_PUBLIC_KEY",
)
).value,
(
await envvars.retrieve(
triggerProjectReference,
environment,
"VAPID_PRIVATE_KEY",
)
).value,
);
for (const subscription of await getUserSubscriptions()) {
const endpoint = subscriptionEndpoint(subscription.subscription);
if (endpoint) {
if (seenEndpoints.has(endpoint)) continue;
seenEndpoints.add(endpoint);
}
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 {};
},
});
|