Integrated notify from legacy triggered by sync.

config/default.cjs

@@ -201,4 +201,8 @@ module.exports = {
 		seenExpiry: 30, // days
 		finalExpiry: 365, // days
 	},
+	email: {
+		// Resend
+		apiKey: null,
+	},
 };

package.json

@@ -57,6 +57,7 @@
     "path-to-regexp": "6.3.0",
     "pg": "^8.22.0",
     "redis": "^6.1.0",
+    "resend": "^6.17.2",
     "sharp": "^0.35.3",
     "sirv": "^3.0.2",
     "template-format": "^1.2.5",

src/alerts.js

@@ -7,6 +7,7 @@ import { getIdsBySlug } from './cache.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();
@@ -393,6 +394,208 @@ export async function updateNotifications(updatedNotification, reqUser) {
 	return updatedCount;
 }
 
+function groupByKey(data, byKey, itemKey) {
+	return data.reduce((acc, item) => {
+		if (!acc[item[byKey]]) {
+			acc[item[byKey]] = [];
+		}
+
+		acc[item[byKey]].push(item[itemKey]);
+
+		return acc;
+	}, {});
+}
+
+export async function notify(sceneIds) {
+	const scenes = await fetchScenesById(sceneIds);
+
+	const [
+		releasesActors,
+		releasesTags,
+		rawAlerts,
+		alertsActors,
+		alertsTags,
+		alertsEntities,
+		alertsMatches,
+		alertsStashes,
+	] = await Promise.all([
+		knex('releases_actors').whereIn('release_id', sceneIds),
+		knex('releases_tags').whereIn('release_id', sceneIds),
+		knex('alerts'),
+		knex('alerts_actors'),
+		knex('alerts_tags'),
+		knex('alerts_entities'),
+		knex('alerts_matches'),
+		knex('alerts_stashes'),
+	]);
+
+	const actorIdsByReleaseId = groupByKey(releasesActors, 'release_id', 'actor_id');
+	const tagIdsByReleaseId = groupByKey(releasesTags, 'release_id', 'tag_id');
+	const alertsActorsByAlertId = groupByKey(alertsActors, 'alert_id', 'actor_id');
+	const alertsTagsByAlertId = groupByKey(alertsTags, 'alert_id', 'tag_id');
+	const alertsEntitiesByAlertId = 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,
+		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,
+		actors: alertsActorsByAlertId[alert.id] || [],
+		tags: alertsTagsByAlertId[alert.id] || [],
+		entities: alertsEntitiesByAlertId[alert.id] || [],
+		matches: alertsMatchesByAlertId[alert.id] || [],
+		stashes: alertsStashesByAlertId[alert.id] || [],
+	}));
+
+	const curatedScenes = scenes
+		.filter((scene) => scene.isNew)
+		.map((scene) => ({
+			id: scene.id,
+			title: scene.title,
+			description: scene.description,
+			actorIds: actorIdsByReleaseId[scene.id] || [],
+			tagIds: tagIdsByReleaseId[scene.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.actors.length === 0
+					&& alert.tags.length === 0
+					&& alert.entities.length === 0
+					&& alert.matches.length === 0) {
+					// we must match on *something*, this is likely a release date alert handled separately
+					return false;
+				}
+
+				if (alert.actors.length > 0 && !alert.actors[alert.allActors ? 'every' : 'some']((actorId) => scene.actorIds.includes(actorId))) {
+					return false;
+				}
+
+				if (alert.tags.length > 0 && !alert.tags[alert.allTags ? 'every' : 'some']((tagId) => scene.tagIds.includes(tagId))) {
+					return false;
+				}
+
+				// multiple entities can only be matched in OR mode
+				if (alert.entities.length > 0 && !alert.entities.some((alertEntityId) => alertEntityId === scene.entityId || alertEntityId === 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.actors[alert.allActors ? 'every' : 'some']((actorId) => scene.actorIds.includes(actorId))) {
+				return true;
+			}
+
+			if (alert.tags.length > 0 && alert.tags[alert.allTags ? 'every' : 'some']((tagId) => scene.tagIds.includes(tagId))) {
+				return true;
+			}
+
+			// multiple entities can only be matched in OR mode
+			if (alert.entities.length > 0 && alert.entities.some((alertEntityId) => alertEntityId === scene.entityId || alertEntityId === 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,
+		}));
+	});
+
+	const notifications = Object.values(Object.fromEntries(triggers // prevent multiple notifications for the same scene
+		.filter((trigger) => trigger.alert.notify)
+		.map((trigger) => [`${trigger.alert.userId}:${trigger.sceneId}`, trigger])))
+		.map((trigger) => ({
+			user_id: trigger.alert.userId,
+			alert_id: trigger.alert.id,
+			scene_id: trigger.sceneId,
+		}));
+
+	const uniqueStashes = Object.values(Object.fromEntries(triggers.flatMap((trigger) => trigger.alert.stashes.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'));
+		}
+	}
+
+	return triggers;
+}
+
 async function alertSceneReleases() {
 	const alerts = await knex('alerts_scenes')
 		.select(
@@ -403,8 +606,8 @@ async function alertSceneReleases() {
 			'releases.slug as scene_slug',
 			'releases.date as scene_date',
 		)
-		.leftJoin('alerts', 'alerts.id', 'alerts_scenes.alert_id')
+		.innerJoin('alerts', 'alerts.id', 'alerts_scenes.alert_id')
-		.leftJoin('releases', 'releases.id', 'alerts_scenes.scene_id')
+		.innerJoin('releases', 'releases.id', 'alerts_scenes.scene_id')
 		.where('alerts.is_expired', false)
 		.where('releases.date', '<=', knex.fn.now());
 

src/sync.js

@@ -5,6 +5,7 @@ import { format } from 'date-fns';
 
 import chunk from '../utils/chunk.js';
 import filterTitle from '../utils/filter-title.js';
+import { notify } from './alerts.js';
 import { knexOwner as knex } from './knex.js';
 import initLogger from './logger.js';
 import { indexApi, searchApi } from './manticore.js';
@@ -277,6 +278,7 @@ export async function syncScenes(releaseIds) {
 	await knex.raw('REFRESH MATERIALIZED VIEW scenes_meta;');
 
 	await syncManticoreScenes(releaseIds);
+	await notify(releaseIds);
 }
 
 export async function syncManticoreMovies(movieIds) {