import config from 'config'; import { CronJob } from 'cron'; import { add } from 'date-fns'; import escapeRegexp from 'escape-string-regexp'; import markdownIt from 'markdown-it'; import { Resend } from 'resend'; import promiseProps from '../utils/promise-props.js'; import { getIdsBySlug } from './cache.js'; import { emailTemplates } from './email-templates.js'; import { HttpError } from './errors.js'; import { knexOwner as knex } from './knex.js'; import initLogger from './logger.js'; import { indexApi } from './manticore.js'; import { fetchScenesById } from './scenes.js'; const logger = initLogger(); const resend = (() => { if (config.email.enabled) { try { return new Resend(config.email.apiKey); } catch (error) { logger.error(`Failed to initialize Resend: ${error.message}`); } } return null; })(); function getImagePath(media) { if (!media) { return null; } if (media.isS3) { return new URL(media.lazy, config.media.s3Path).href; } return new URL(`${config.media.mediaPath}/${media.lazy}`, config.web.address).href; } const md = markdownIt({ html: true, }); md.disable('code'); function curateAlert(alert, context = {}) { return { id: alert.id, notify: alert.notify, email: alert.email, createdAt: alert.created_at, isFromPreset: alert.from_preset, and: { fields: alert.all, actors: alert.all_actors, tags: alert.all_tags, entities: alert.all_entities, matches: alert.all_tags, }, actors: context.actors?.map((actor) => ({ id: actor.actor_id, name: actor.actor_name, slug: actor.actor_slug, })) || [], tags: context.tags?.map((tag) => ({ id: tag.tag_id, name: tag.tag_name, slug: tag.tag_slug, })) || [], entities: context.entities?.map((entity) => ({ id: entity.entity_id, name: entity.entity_name, slug: entity.entity_slug, type: entity.entity_type, })) || [], matches: context.matches?.map((match) => ({ id: match.id, property: match.property, expression: match.expression, })) || [], scenes: context.scenes?.map((scene) => ({ id: scene.scene_id, title: scene.scene_title, slug: scene.scene_slug, date: scene.scene_date, })) || [], stashes: context.stashes?.map((stash) => ({ id: stash.stash_id, name: stash.stash_name, slug: stash.stash_slug, isPrimary: stash.stash_primary, })) || [], comment: alert.comment, meta: alert.meta, }; } export async function fetchAlerts(user, alertIds) { const { alerts, actors, tags, entities, scenes, matches, stashes, } = await promiseProps({ alerts: knex('alerts') .where('user_id', user.id) .where('is_expired', false) .where((builder) => { if (alertIds) { builder.whereIn('id', alertIds); } }) .orderBy('created_at', 'desc'), actors: knex('alerts_actors') .select('alerts_actors.*', 'actors.name as actor_name', 'actors.slug as actor_slug') .leftJoin('alerts', 'alerts.id', 'alerts_actors.alert_id') .leftJoin('actors', 'actors.id', 'alerts_actors.actor_id') .where('alerts.user_id', user.id) .where('alerts.is_expired', false) .where((builder) => { if (alertIds) { builder.whereIn('alerts.id', alertIds); } }), tags: knex('alerts_tags') .select('alerts_tags.*', 'tags.name as tag_name', 'tags.slug as tag_slug') .leftJoin('alerts', 'alerts.id', 'alerts_tags.alert_id') .leftJoin('tags', 'tags.id', 'alerts_tags.tag_id') .where('alerts.user_id', user.id) .where('alerts.is_expired', false) .where((builder) => { if (alertIds) { builder.whereIn('alerts.id', alertIds); } }), entities: knex('alerts_entities') .select('alerts_entities.*', 'entities.name as entity_name', 'entities.slug as entity_slug', 'entities.type as entity_type') .leftJoin('alerts', 'alerts.id', 'alerts_entities.alert_id') .leftJoin('entities', 'entities.id', 'alerts_entities.entity_id') .where('alerts.user_id', user.id) .where('alerts.is_expired', false) .where((builder) => { if (alertIds) { builder.whereIn('alerts.id', alertIds); } }), scenes: knex('alerts_scenes') .select('alerts_scenes.*', 'releases.id as scene_id', 'releases.slug as scene_slug', 'releases.title as scene_title', 'releases.date as scene_date') .leftJoin('alerts', 'alerts.id', 'alerts_scenes.alert_id') .leftJoin('releases', 'releases.id', 'alerts_scenes.scene_id') .where('alerts.user_id', user.id) .where('alerts.is_expired', false) .where((builder) => { if (alertIds) { builder.whereIn('alerts.id', alertIds); } }), matches: knex('alerts_matches') .select('alerts_matches.*') .leftJoin('alerts', 'alerts.id', 'alerts_matches.alert_id') .where('alerts.user_id', user.id) .where('alerts.is_expired', false) .where((builder) => { if (alertIds) { builder.whereIn('alerts.id', alertIds); } }), stashes: knex('alerts_stashes') .select('alerts_stashes.*', 'stashes.id as stash_id', 'stashes.name as stash_name', 'stashes.slug as stash_slug', 'stashes.primary as stash_primary') .leftJoin('alerts', 'alerts.id', 'alerts_stashes.alert_id') .leftJoin('stashes', 'stashes.id', 'alerts_stashes.stash_id') .where('alerts.user_id', user.id) .where('alerts.is_expired', false) .where((builder) => { if (alertIds) { builder.whereIn('alerts.id', alertIds); } }), }); const curatedAlerts = alerts.map((alert) => curateAlert(alert, { actors: actors.filter((actor) => actor.alert_id === alert.id), tags: tags.filter((tag) => tag.alert_id === alert.id), entities: entities.filter((entity) => entity.alert_id === alert.id), matches: matches.filter((match) => match.alert_id === alert.id), scenes: scenes.filter((scene) => scene.alert_id === alert.id), stashes: stashes.filter((stash) => stash.alert_id === alert.id), })); return curatedAlerts; } export async function createAlert(alert, reqUser) { if (!reqUser) { throw new HttpError('You are not authenthicated', 401); } const tagIds = (await getIdsBySlug(alert.tags, 'tags') || []).concat(alert.tagIds || []); const entityIds = (await getIdsBySlug(alert.entities || [], 'entities')).concat(alert.entityIds || []); const actorIds = [...(alert.actors || []), ...(alert.actorIds || [])]; // for consistency with tagIds and entityIds const stashIds = [...(alert.stashes || []), ...(alert.stashIds || [])]; // for consistency with tagIds and entityIds const sceneIds = [...(alert.scenes || []), ...(alert.sceneIds || [])]; // for consistency with tagIds and entityIds if (actorIds.length === 0 && tagIds.length === 0 && entityIds.length === 0 && (!alert.matches || alert.matches.length === 0) && sceneIds.length === 0) { throw new HttpError('Alert must contain at least one actor, tag, entity or scene', 400); } if (alert.matches?.some((match) => !match.property || !match.expression)) { throw new HttpError('Match must define a property and an expression', 400); } if (tagIds.length < (alert.tags?.length || 0) + (alert.tagIds?.length || 0)) { throw new HttpError('Failed to resolve all tags'); } if (entityIds.length < (alert.entities?.length || 0) + (alert.entityIds?.length || 0)) { throw new HttpError('Failed to resolve all entities'); } const [{ id: alertId }] = await knex('alerts') .insert({ user_id: reqUser.id, notify: alert.notify, email: alert.email, all: alert.all, all_actors: alert.allActors, all_entities: alert.allEntities, all_tags: alert.allTags, all_matches: alert.allMatches, from_preset: alert.preset, comment: alert.comment, meta: alert.meta, }) .returning('id'); await Promise.all([ actorIds?.length > 0 && knex('alerts_actors').insert(actorIds.map((actorId) => ({ alert_id: alertId, actor_id: actorId, }))), tagIds?.length > 0 && knex('alerts_tags').insert(tagIds.map((tagId) => ({ alert_id: alertId, tag_id: tagId, }))), alert.matches?.length > 0 && knex('alerts_matches').insert(alert.matches.map((match) => ({ alert_id: alertId, property: match.property, expression: /\/.*\//.test(match.expression) ? match.expression.slice(1, -1) : escapeRegexp(match.expression), }))), stashIds?.length > 0 && knex('alerts_stashes').insert(stashIds.map((stashId) => ({ alert_id: alertId, stash_id: stashId, }))), entityIds?.length > 0 && knex('alerts_entities').insert(entityIds.map((entityId) => ({ alert_id: alertId, entity_id: entityId, })).slice(0, alert.allEntities ? 1 : Infinity)), // one scene can never match multiple entities in AND mode sceneIds?.length > 0 && knex('alerts_scenes').insert(sceneIds.map((sceneId) => ({ alert_id: alertId, scene_id: sceneId, }))), ]); await Promise.all([ actorIds?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_actors'), tagIds?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_tags'), entityIds?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_entities'), sceneIds?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_scenes'), ]); const [newAlert] = await fetchAlerts(reqUser, [alertId]); return newAlert; } export async function removeAlert(alertId, reqUser) { const alert = await knex('alerts') .where('alerts.id', alertId) .where('alerts.user_id', reqUser.id) .select( 'alerts.id', knex.raw('coalesce(array_agg(distinct alerts_actors.actor_id) filter (where alerts_actors.actor_id is not null), \'{}\') as actor_ids'), knex.raw('coalesce(array_agg(distinct alerts_entities.entity_id) filter (where alerts_entities.entity_id is not null), \'{}\') as entity_ids'), knex.raw('coalesce(array_agg(distinct alerts_tags.tag_id) filter (where alerts_tags.tag_id is not null), \'{}\') as tag_ids'), knex.raw('coalesce(array_agg(distinct alerts_scenes.scene_id) filter (where alerts_scenes.scene_id is not null), \'{}\') as scene_ids'), ) .leftJoin('alerts_actors', 'alerts_actors.alert_id', 'alerts.id') .leftJoin('alerts_entities', 'alerts_entities.alert_id', 'alerts.id') .leftJoin('alerts_tags', 'alerts_tags.alert_id', 'alerts.id') .leftJoin('alerts_matches', 'alerts_matches.alert_id', 'alerts.id') .leftJoin('alerts_scenes', 'alerts_scenes.alert_id', 'alerts.id') .groupBy('alerts.id') .first(); if (!alert) { throw new HttpError(`Could not find alert ${alertId}`, 404); } const [removed] = await knex('alerts') .where('id', alertId) .where('user_id', reqUser.id) .delete() .returning('*'); if (!removed) { throw new HttpError(`Could not remove alert ${alertId}`, 404); } // slow not critical for response, don't await Promise.all([ alert.actor_ids.length > 0 && knex.schema.refreshMaterializedView('alerts_users_actors'), alert.tag_ids.length > 0 && knex.schema.refreshMaterializedView('alerts_users_tags'), alert.entity_ids?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_entities'), alert?.scene_ids?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_scenes'), ]); return curateAlert(removed); } function curateNotification(notification, scenes) { const scene = scenes.find((sceneX) => sceneX.id === notification.scene_id); return { id: notification.id, sceneId: notification.scene_id, scene, alertId: notification.alert_id, matchedSceneIds: notification.alert_scenes, matchedActors: scene.actors.filter((actor) => notification.alert_actors.includes(actor.id)), matchedTags: scene.tags.filter((tag) => notification.alert_tags.includes(tag.id)), matchedEntity: [scene.channel, scene.network].find((entity) => notification.alert_entities.includes(entity?.id)) || null, matchedExpressions: notification.alert_matches .filter((match) => new RegExp(match.expression, 'iu').test(scene[match.property])) .map((match) => ({ id: match.id, property: match.property, expression: match.expression, })), isSeen: notification.seen, createdAt: notification.created_at, }; } export async function fetchUnseenNotificationsCount(reqUser) { if (!reqUser) { return null; } const rawUnseen = await knex('notifications') .select(knex.raw('count(id)')) .where('user_id', reqUser.id) .where('seen', false) .first(); return Number(rawUnseen.count); } export async function fetchNotifications(reqUser, options = {}) { if (!reqUser) { return []; } const [notifications, unseen] = await Promise.all([ knex('notifications') .select( 'notifications.*', 'alerts.id as alert_id', knex.raw('coalesce(array_agg(alerts_actors.actor_id) filter (where alerts_actors.id is not null), \'{}\') as alert_actors'), knex.raw('coalesce(array_agg(alerts_tags.tag_id) filter (where alerts_tags.id is not null), \'{}\') as alert_tags'), knex.raw('coalesce(array_agg(alerts_entities.entity_id) filter (where alerts_entities.id is not null), \'{}\') as alert_entities'), knex.raw('coalesce(array_agg(alerts_scenes.scene_id) filter (where alerts_scenes.id is not null), \'{}\') as alert_scenes'), knex.raw('coalesce(json_agg(alerts_matches) filter (where alerts_matches.id is not null), \'[]\') as alert_matches'), ) .leftJoin('alerts', 'alerts.id', 'notifications.alert_id') .leftJoin('alerts_actors', 'alerts_actors.alert_id', 'alerts.id') .leftJoin('alerts_tags', 'alerts_tags.alert_id', 'alerts.id') .leftJoin('alerts_entities', 'alerts_entities.alert_id', 'alerts.id') .leftJoin('alerts_matches', 'alerts_matches.alert_id', 'alerts.id') .leftJoin('alerts_scenes', 'alerts_scenes.alert_id', 'alerts.id') .where('notifications.user_id', reqUser.id) .where('notifications.notify', true) // notification table is also used to defer e-mail alerts .limit(options.limit) .groupBy('notifications.id', 'alerts.id') .orderBy('created_at', 'desc'), fetchUnseenNotificationsCount(reqUser), ]); const scenes = await fetchScenesById(notifications.map((notification) => notification.scene_id)); const curatedNotifications = notifications.map((notification) => curateNotification(notification, scenes)); return { notifications: curatedNotifications, unseen, }; } export async function updateNotification(notificationId, updatedNotification, reqUser) { const [updated] = await knex('notifications') .where('id', notificationId) .where('user_id', reqUser.id) .update({ seen: updatedNotification.seen, }) .returning('*'); if (!updated) { throw new HttpError(`No notification ${notificationId} found to update`, 404); } return updated.id; } export async function updateNotifications(updatedNotification, reqUser) { const updatedCount = await knex('notifications') .where('user_id', reqUser.id) .update({ seen: updatedNotification.seen, seen_at: updatedNotification.seen ? knex.fn.now() : null, }); return updatedCount; } function groupByKey(data, byKey, itemKey) { return data.reduce((acc, item) => { if (!acc[item[byKey]]) { acc[item[byKey]] = []; } acc[item[byKey]].push(itemKey ? item[itemKey] : item); return acc; }, {}); } const frequencies = { daily: { days: 1 }, weekly: { weeks: 1 }, monthly: { months: 1 }, }; async function sendEmailAlerts() { if (!config.email.enabled) { return; } if (!resend) { logger.warn('E-mails are enabled, but Resend is not initialized'); return; } const qualifyingUsers = await knex('notifications') .select('users.id as user_id', 'users.settings', knex.raw('max(emailed_at) as last_email')) .where('notifications.email', true) .innerJoin('users', 'users.id', 'notifications.user_id') .groupBy('users.id') .having(knex.raw('bool_or(emailed_at is null)')); const dueUserIds = qualifyingUsers .filter(({ settings, last_email }) => { const frequency = frequencies[settings?.alertEmailFrequency]; if (!frequency || !last_email) { // email setting set to immediate or invalid, or no recent emails sent return true; } // is this last e-mail sent before the frequency setting timeframe? return add(last_email, frequency) <= new Date(); }) .map((row) => row.user_id); // immediately mark as e-mailed to prevent double e-mails if cronjob coincides with direct trigger const markedNotifications = await knex('notifications') .where('email', true) .whereNull('emailed_at') .whereIn('user_id', dueUserIds) .update('emailed_at', knex.fn.now()) .returning('*'); const userIds = Array.from(new Set(markedNotifications.map((notification) => notification.user_id))); const sceneIds = Array.from(new Set(markedNotifications.map((notification) => notification.scene_id))); const alertIds = Array.from(new Set(markedNotifications.map((notification) => notification.alert_id))); const scenes = await fetchScenesById(sceneIds); const [ users, alertsActors, alertsTags, alertsEntities, alertsScenes, alertsMatches, ] = await Promise.all([ knex('users') .select('users.id', 'users.username', 'users.email', 'users.settings') .whereIn('users.id', userIds), knex('alerts_actors') .select('alerts_actors.alert_id', 'actors.id', 'actors.name', 'actors.slug') .whereIn('alerts_actors.alert_id', alertIds) .leftJoin('actors', 'actors.id', 'alerts_actors.actor_id'), knex('alerts_tags') .select('alerts_tags.alert_id', 'tags.id', 'tags.name', 'tags.slug') .whereIn('alerts_tags.alert_id', alertIds) .leftJoin('tags', 'tags.id', 'alerts_tags.tag_id'), knex('alerts_entities') .select('alerts_entities.alert_id', 'entities.id', 'entities.type', 'entities.name', 'entities.slug') .whereIn('alerts_entities.alert_id', alertIds) .leftJoin('entities', 'entities.id', 'alerts_entities.entity_id'), knex('alerts_scenes') .whereIn('alerts_scenes.alert_id', alertIds), knex('alerts_matches') .whereIn('alerts_matches.alert_id', alertIds), ]); const usersByUserId = Object.fromEntries(users.map((user) => [user.id, user])); const scenesBySceneId = Object.fromEntries(scenes.map((scene) => [scene.id, scene])); const alertActorsByAlertId = groupByKey(alertsActors, 'alert_id'); const alertTagsByAlertId = groupByKey(alertsTags, 'alert_id'); const alertEntitiesByAlertId = groupByKey(alertsEntities, 'alert_id'); const alertMatchesByAlertId = groupByKey(alertsMatches, 'alert_id'); const alertSceneIdsByAlertId = groupByKey(alertsScenes, 'alert_id', 'scene_id'); const triggers = markedNotifications .map((notification) => ({ notificationId: notification.id, scene: scenesBySceneId[notification.scene_id] || null, posterUrl: getImagePath(scenesBySceneId[notification.scene_id]?.poster), user: usersByUserId[notification.user_id] || null, alertId: notification.alert_id, alertTags: alertTagsByAlertId[notification.alert_id] || [], alertActors: alertActorsByAlertId[notification.alert_id] || [], alertEntities: alertEntitiesByAlertId[notification.alert_id] || [], alertMatches: alertMatchesByAlertId[notification.alert_id] || [], alertSceneIds: alertSceneIdsByAlertId[notification.alert_id] || [], })) .filter((notification) => !!notification.user && !!notification.scene && notification.user.settings?.alertEmailsEnabled !== false); // keep the e-mails marked as 'sent', if emails get re-enabled there shouldn't be a barrage of old notifications const emails = await Promise.all(Object.values(usersByUserId).map(async (user) => { const userTriggers = Object.values(Object.fromEntries(triggers .filter((trigger) => trigger.user.id === user.id) .map((trigger) => [trigger.scene.id, trigger]))) .toSorted((triggerA, triggerB) => triggerA.scene.effectiveDate - triggerB.scene.effectiveDate); if (userTriggers.length === 0) { return null; } try { const { html, text } = await emailTemplates.render('alert.vue', { props: { user, triggers: userTriggers, }, }); return { from: config.email.from, to: user.email, subject: `Notification for ${userTriggers.length} scenes on traxxx`, html, text, }; } catch (error) { logger.warn(`Failed to render e-mail for '${user.username}' with scenes ${userTriggers.map((trigger) => `${trigger.notificationId}: ${trigger.scene.id}`)}: ${error.message}`); return null; } })).then((allEmails) => allEmails.filter(Boolean)); if (emails.length === 0) { logger.info(`No alert e-mails sent`); return; } const { data, error } = await resend.batch.send(emails); if (error) { logger.error(`Failed to send ${emails.length} alert e-mails (${error.statusCode} ${error.name}): ${error.message}`); } else { logger.info(`Sent ${data.data.length}/${emails.length} alert e-mails`); } } export async function notify(sceneIds, options = {}) { const scenes = await fetchScenesById(sceneIds); const [ rawAlerts, alertsActors, alertsTags, alertsEntities, alertsMatches, alertsStashes, ] = await Promise.all([ knex('alerts') .select('alerts.*', 'users.username', 'users.email as user_email', 'users.settings as user_settings') .leftJoin('users', 'users.id', 'alerts.user_id'), knex('alerts_actors').select('alert_id', 'actor_id'), knex('alerts_tags').select('alert_id', 'tag_id'), knex('alerts_entities').select('alert_id', 'entity_id'), knex('alerts_matches'), knex('alerts_stashes'), ]); const alertActorIdsByAlertId = groupByKey(alertsActors, 'alert_id', 'actor_id'); const alertTagIdsByAlertId = groupByKey(alertsTags, 'alert_id', 'tag_id'); const alertEntityIdsByAlertId = groupByKey(alertsEntities, 'alert_id', 'entity_id'); const alertsStashesByAlertId = groupByKey(alertsStashes, 'alert_id', 'stash_id'); const alertsMatchesByAlertId = alertsMatches.reduce((acc, alertMatch) => { if (!acc[alertMatch.alert_id]) { acc[alertMatch.alert_id] = []; } acc[alertMatch.alert_id].push({ property: alertMatch.property, expression: new RegExp(alertMatch.expression, 'iu'), }); return acc; }, {}); const alerts = rawAlerts .map((alert) => ({ id: alert.id, userId: alert.user_id, username: alert.username, userEmail: alert.user_email, userSettings: alert.user_settings, notify: alert.notify, email: alert.email, all: alert.all, allActors: alert.all_actors, allEntities: alert.all_entities, allTags: alert.all_tags, allMatches: alert.all_matches, actorIds: alertActorIdsByAlertId[alert.id] || [], tagIds: alertTagIdsByAlertId[alert.id] || [], entityIds: alertEntityIdsByAlertId[alert.id] || [], matches: alertsMatchesByAlertId[alert.id] || [], stashIds: alertsStashesByAlertId[alert.id] || [], })) .filter((alert) => options.spoofUserIds ? options.spoofUserIds.includes(alert.userId) : true); const curatedScenes = scenes .filter((scene) => scene.isNew || options.spoofNew) .toSorted((sceneA, sceneB) => sceneB.date - sceneA.date) .map((scene) => ({ id: scene.id, title: scene.title, description: scene.description, actorIds: scene.actors.map((actor) => actor.id), tagIds: scene.tags.map((tag) => tag.id), entityId: scene.channel?.id || scene.network?.id || null, parentEntityId: scene.network?.id, })); const triggers = alerts.flatMap((alert) => { const alertScenes = curatedScenes.filter((scene) => { if (alert.all) { if (alert.actorIds.length === 0 && alert.tagIds.length === 0 && alert.entityIds.length === 0 && alert.matches.length === 0) { // we must match on *something*, this is likely a release date alert handled separately return false; } if (alert.actorIds.length > 0 && !alert.actorIds[alert.allActors ? 'every' : 'some']((actorId) => scene.actorIds.includes(actorId))) { return false; } if (alert.tagIds.length > 0 && !alert.tagIds[alert.allTags ? 'every' : 'some']((tagId) => scene.tagIds.includes(tagId))) { return false; } // multiple entities can only be matched in OR mode if (alert.entityIds.length > 0 && !alert.entityIds.some((entityId) => entityId === scene.entityId || entityId === scene.parentEntityId)) { return false; } if (alert.matches.length > 0 && !alert.matches[alert.allMatches ? 'every' : 'some']((match) => match.expression.test(scene[match.property]))) { return false; } return true; } if (alert.matches.length > 0 && alert.actorIds[alert.allActors ? 'every' : 'some']((actorId) => scene.actorIds.includes(actorId))) { return true; } if (alert.tagIds.length > 0 && alert.tagIds[alert.allTags ? 'every' : 'some']((tagId) => scene.tagIds.includes(tagId))) { return true; } // multiple entities can only be matched in OR mode if (alert.entityIds.length > 0 && alert.entityIds.some((entityId) => entityId === scene.entityId || entityId === scene.parentEntityId)) { return true; } if (alert.matches.length > 0 && alert.matches[alert.allMatches ? 'every' : 'some']((match) => match.expression.test(scene[match.property]))) { return true; } return false; }); return alertScenes.map((scene) => ({ sceneId: scene.id, alert, })); }); // e-mails are stored as notifications to facilitate the e-mail frequency buffer const notifications = Object.values(triggers // prevent multiple notifications for the same scene .map((trigger) => ({ ...trigger, notify: trigger.alert.notify && trigger.alert.userSettings?.alertNotificationsEnabled !== false, email: trigger.alert.email && trigger.alert.userEmail && trigger.alert.userSettings?.alertEmailsEnabled !== false, })) .filter((trigger) => trigger.notify || trigger.email) .reduce((acc, trigger) => { const key = `${trigger.alert.userId}:${trigger.sceneId}`; acc[key] = { ...trigger, notify: acc[key]?.notify || trigger.notify, email: acc[key]?.email || trigger.email, }; return acc; }, {})) .map((trigger) => ({ user_id: trigger.alert.userId, alert_id: trigger.alert.id, scene_id: trigger.sceneId, // we assign the alert outputs to this specific notification, instead of joining them from the alerts table // this ensures changing the alert outputs later doesn't trigger a barrage of e-mails from old notifications notify: trigger.notify, email: trigger.email, })); if (notifications.length === 0) { logger.verbose(`No alert notifications triggered`); } const uniqueStashes = Object.values(Object.fromEntries(triggers.flatMap((trigger) => trigger.alert.stashIds.map((stashId) => ({ stashId, sceneId: trigger.sceneId, userId: trigger.alert.userId, }))).map((stash) => [`${stash.stashId}:${stash.sceneId}`, stash]))); const stashEntries = uniqueStashes.map((stash) => ({ scene_id: stash.sceneId, stash_id: stash.stashId, })); const [stashed] = await Promise.all([ stashEntries.length > 0 ? knex('stashes_scenes') .insert(stashEntries) .onConflict(['stash_id', 'scene_id']) .ignore() : [], notifications.length > 0 ? knex('notifications') .insert(notifications) : [], ]); if (stashed.length > 0) { // we need created_at from the databased, but user_id is not returned. it's easier to query it than to try and merge it with the input data const stashedEntries = await knex('stashes_scenes') .select('stashes_scenes.*', 'stashes.user_id') .leftJoin('stashes', 'stashes.id', 'stashes_scenes.stash_id') .whereIn('stashes_scenes.id', stashed.map((stash) => stash.id)); const docs = stashedEntries.map((stash) => ({ replace: { index: 'scenes_stashed', id: stash.id, doc: { scene_id: stash.scene_id, user_id: stash.user_id, stash_id: stash.stash_id, created_at: Math.round(stash.created_at.getTime() / 1000), }, }, })); if (docs.length > 0) { await indexApi.bulk(docs.map((doc) => JSON.stringify(doc)).join('\n')); } } await sendEmailAlerts(); return triggers; } async function alertSceneReleases() { const alerts = await knex('alerts_scenes') .select( 'alerts.*', 'releases.id as scene_id', 'releases.title as scene_title', 'releases.slug as scene_slug', 'releases.date as scene_date', ) .innerJoin('alerts', 'alerts.id', 'alerts_scenes.alert_id') .innerJoin('releases', 'releases.id', 'alerts_scenes.scene_id') .where('alerts.is_expired', false) .where('releases.date', '<=', knex.fn.now()); if (alerts.length === 0) { return; } const trx = await knex.transaction(); try { await trx('notifications') .insert(alerts.map((alert) => ({ user_id: alert.user_id, scene_id: alert.scene_id, alert_id: alert.id, notify: alert.notify, email: alert.email, }))); await trx('alerts') .whereIn('id', alerts.map((alert) => alert.id)) .update('is_expired', true); await trx.commit(); await knex.schema.refreshMaterializedView('alerts_users_scenes'); } catch (error) { await trx.rollback(); logger.error(`Failed to notify alerts: ${error.message}`); } } async function purgeExpiredAlerts() { const purgedNotifications = await knex('notifications') .where((builder) => { builder .where('seen', true) .where((seenBuilder) => { seenBuilder .whereNull('seen_at') // marked seen before seen_at was added .orWhereRaw('seen_at < now() - (?::int * interval \'1 day\')', [config.alerts.seenExpiry]); // '? day' would be interpreted literally }); }) .orWhereRaw('created_at < now() - (?::int * interval \'1 day\')', [config.alerts.finalExpiry]) // don't accumulate notifications indefinitely .delete(); const purgedAlerts = await knex('alerts') .where('is_expired', true) .whereNotExists( knex('notifications') .whereRaw('notifications.alert_id = alerts.id'), ) .delete(); await knex.schema.refreshMaterializedView('alerts_users_scenes'); logger.info(`Purged ${purgedNotifications} notifications and ${purgedAlerts} alerts`); } export function initAlertCron() { CronJob.from({ cronTime: config.alerts.crontab, async onTick() { await Promise.all([ alertSceneReleases(), purgeExpiredAlerts(), ]); // allow release alerts to set e-mail notifications await sendEmailAlerts(); }, start: config.alerts.enabled, runOnInit: true, }); }