File size: 1,738 Bytes
57aa51e | 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 | import {
condition,
continueAsNew,
proxyActivities,
setHandler,
sleep,
} from '@temporalio/workflow';
import { Email, emailSignal } from '@gitroom/orchestrator/signals/email.signal';
import { EmailActivity } from '@gitroom/orchestrator/activities/email.activity';
const { getUserOrgs, sendEmailAsync } = proxyActivities<EmailActivity>({
startToCloseTimeout: '10 minute',
taskQueue: 'main',
cancellationType: 'ABANDON',
retry: {
maximumAttempts: 3,
backoffCoefficient: 1,
initialInterval: '2 minutes',
},
});
export async function digestEmailWorkflow({
organizationId,
queue = [],
}: {
organizationId: string;
queue?: Email[];
}) {
setHandler(emailSignal, (data) => {
queue.push(...data);
});
while (true) {
await condition(() => queue.length > 0);
await sleep(3600000);
// Take a snapshot batch and immediately clear queue.
const batch = queue.splice(0, queue.length);
queue = [];
const org = await getUserOrgs(organizationId);
for (const user of org.users) {
const allowFailure = user.user.sendFailureEmails ? 'fail' : null;
const allowSuccess = user.user.sendSuccessEmails ? 'success' : null;
const toSend = batch.filter(
(email) =>
email.type === allowFailure ||
email.type === allowSuccess ||
email.type === 'info'
);
if (toSend.length === 0) continue;
await sendEmailAsync(
user.user.email,
toSend.length === 1
? toSend[0].title
: `[Postiz] Your latest notifications`,
toSend.map((p) => p.message).join('<br/>'),
'bottom'
);
}
return await continueAsNew({
organizationId,
queue,
});
}
}
|