Purging expired notifications and alerts. Fixed manticore stash sync.

components/alerts/alerts.vue

@@ -281,6 +281,22 @@ function toggleFilterCombined() {
 							</span>
 						</span>
 					</span>
+
+					<span
+						v-for="scene in alert.scenes"
+						:key="`alert-${alert.id}-${scene.id}`"
+						class="alert-scene"
+					>
+						<time
+							:datetime="scene.date.toISOString()"
+							class="alert-date"
+						>{{ format(scene.date, 'yyyy-MM-dd') }}</time>
+
+						<a
+							:href="`/scene/${scene.id}/${scene.slug}`"
+							class="alert-value alert-title ellipsis link"
+						>{{ scene.title }}</a>
+					</span>
 				</div>
 
 				<div class="alert-meta">
@@ -476,6 +492,16 @@ function toggleFilterCombined() {
 	display: flex;
 }
 
+.alert-scene {
+	display: flex;
+	gap: .5rem;
+	padding: 0 .5rem;
+}
+
+.alert-date {
+	flex-shrink: 0;
+}
+
 .alert-actions {
 	display: flex;
 	align-items: center;

components/header/notifications.vue

@@ -108,7 +108,15 @@ onMounted(async () => {
 					class="notif-body notif-link nolink"
 				>
 					<span class="notif-trigger">
-						<span class="notif-details">
+						<span
+							v-if="notif.matchedSceneIds.length > 0"
+							class="notif-details"
+						>Scene is released</span>
+
+						<span
+							v-else
+							class="notif-details"
+						>
 							New
 
 							<template

components/stashes/heart.vue

@@ -22,6 +22,10 @@ const props = defineProps({
 		type: Boolean,
 		default: true,
 	},
+	showAlert: {
+		type: Boolean,
+		default: true,
+	},
 });
 
 const emit = defineEmits(['stashed', 'unstashed']);
@@ -118,6 +122,7 @@ async function alertItem() {
 			...(props.domain === 'actors' && { actors: [props.item.id] }),
 			...(props.domain === 'tags' && { tags: [props.item.id] }),
 			...(props.domain === 'entities' && { entities: [props.item.id] }),
+			...(props.domain === 'scenes' && { scenes: [props.item.id] }),
 			notify: true,
 			email: false,
 			preset: true,
@@ -205,7 +210,7 @@ async function reloadStashes(newStash) {
 		class="bookmarks"
 	>
 		<div
-			v-if="showSecondary && !!itemAlerts"
+			v-if="showSecondary && showAlert && !!itemAlerts"
 			class="alert-trigger"
 		>
 			<Icon

config/default.cjs

@@ -196,4 +196,9 @@ module.exports = {
 		enabled: true,
 		crontab: '*/10 * * * *', // every 10 minutes
 	},
+	alerts: {
+		crontab: '*/10 * * * *', // every 10 minutes
+		seenExpiry: 30, // days
+		finalExpiry: 365, // days
+	},
 };

pages/scene/+Page.vue

@@ -363,6 +363,8 @@ function copySummary() {
 					<Heart
 						domain="scenes"
 						:item="scene"
+						:show-secondary="true"
+						:show-alert="scene.isUpcoming"
 					/>
 
 					<div class="view">

src/alerts.js

@@ -1,11 +1,16 @@
+import config from 'config';
+import { CronJob } from 'cron';
 import escapeRegexp from 'escape-string-regexp';
 
 import promiseProps from '../utils/promise-props.js';
 import { getIdsBySlug } from './cache.js';
 import { HttpError } from './errors.js';
 import { knexOwner as knex } from './knex.js';
+import initLogger from './logger.js';
 import { fetchScenesById } from './scenes.js';
 
+const logger = initLogger();
+
 function curateAlert(alert, context = {}) {
 	return {
 		id: alert.id,
@@ -41,6 +46,12 @@ function curateAlert(alert, context = {}) {
 			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,
@@ -58,11 +69,13 @@ export async function fetchAlerts(user, alertIds) {
 		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);
@@ -74,6 +87,7 @@ export async function fetchAlerts(user, alertIds) {
 			.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);
@@ -84,6 +98,7 @@ export async function fetchAlerts(user, alertIds) {
 			.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);
@@ -94,6 +109,18 @@ export async function fetchAlerts(user, alertIds) {
 			.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);
@@ -103,6 +130,7 @@ export async function fetchAlerts(user, alertIds) {
 			.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);
@@ -113,6 +141,7 @@ export async function fetchAlerts(user, alertIds) {
 			.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);
@@ -125,6 +154,7 @@ export async function fetchAlerts(user, alertIds) {
 		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),
 	}));
 
@@ -140,9 +170,10 @@ export async function createAlert(alert, reqUser) {
 	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)) {
+	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 or entity', 400);
+		throw new HttpError('Alert must contain at least one actor, tag, entity or scene', 400);
 	}
 
 	if (alert.matches?.some((match) => !match.property || !match.expression)) {
@@ -197,12 +228,17 @@ export async function createAlert(alert, reqUser) {
 			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([
-		alert.actors?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_actors'),
+		actorIds?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_actors'),
-		alert.tags?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_tags'),
+		tagIds?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_tags'),
-		alert.entities?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_entities'),
+		entityIds?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_entities'),
+		sceneIds?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_scenes'),
 	]);
 
 	const [newAlert] = await fetchAlerts(reqUser, [alertId]);
@@ -219,11 +255,13 @@ export async function removeAlert(alertId, reqUser) {
 			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();
 
@@ -246,6 +284,7 @@ export async function removeAlert(alertId, reqUser) {
 		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);
@@ -259,6 +298,7 @@ function curateNotification(notification, scenes) {
 		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,
@@ -301,6 +341,7 @@ export async function fetchNotifications(reqUser, options = {}) {
 				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')
@@ -308,6 +349,7 @@ export async function fetchNotifications(reqUser, options = {}) {
 			.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)
 			.limit(options.limit)
 			.groupBy('notifications.id', 'alerts.id')
@@ -345,7 +387,89 @@ export async function updateNotifications(updatedNotification, reqUser) {
 		.where('user_id', reqUser.id)
 		.update({
 			seen: updatedNotification.seen,
+			seen_at: updatedNotification.seen ? knex.fn.now() : null,
 		});
 
 	return updatedCount;
 }
+
+async function alertSceneReleases() {
+	const alerts = await knex('alerts_scenes')
+		.select(
+			'alerts.id',
+			'alerts.user_id',
+			'releases.id as scene_id',
+			'releases.title as scene_title',
+			'releases.slug as scene_slug',
+			'releases.date as scene_date',
+		)
+		.leftJoin('alerts', 'alerts.id', 'alerts_scenes.alert_id')
+		.leftJoin('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,
+			})));
+
+		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() {
+	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() - interval \'? days\'', [config.alerts.seenExpiry]);
+				});
+		})
+		.orWhereRaw('created_at < now() - interval \'? days\'', [config.alerts.finalExpiry]) // don't accumulate notifications indefinitely
+		.delete();
+
+	await knex('alerts')
+		.where('is_expired', true)
+		.whereNotExists(
+			knex('notifications')
+				.whereRaw('notifications.alert_id = alerts.id'),
+		)
+		.delete();
+
+	await knex.schema.refreshMaterializedView('alerts_users_scenes');
+}
+
+export function initAlertCron() {
+	CronJob.from({
+		cronTime: config.alerts.crontab,
+		async onTick() {
+			await Promise.all([
+				alertSceneReleases(),
+				purgeExpiredAlerts(),
+			]);
+		},
+		start: config.alerts.enabled,
+		runOnInit: true,
+	});
+}

src/app.js

@@ -1,3 +1,4 @@
+import { initAlertCron } from './alerts.js';
 import { initCaches } from './cache.js';
 import { initSyncCron } from './sync.js';
 import initServer from './web/server.js';
@@ -7,6 +8,7 @@ async function init() {
 
 	initServer();
 	initSyncCron();
+	initAlertCron();
 }
 
 init();

src/scenes.js

@@ -36,6 +36,7 @@ function curateScene(rawScene, assets, reqUser, context) {
 		entryId: rawScene.entry_id,
 		date: rawScene.date,
 		datePrecision: rawScene.date_precision,
+		isUpcoming: rawScene.date && rawScene.date - new Date() > 0,
 		createdAt: rawScene.created_at,
 		effectiveDate: rawScene.effective_date,
 		description: censor(rawScene.description, context.restriction),
@@ -131,7 +132,6 @@ function curateScene(rawScene, assets, reqUser, context) {
 		poster: curateMedia(assets.poster, { type: 'poster' }),
 		photos: assets.photos?.map((photo) => curateMedia(photo, { type: 'photo' })) || [],
 		caps: assets.caps?.map((cap) => curateMedia(cap, { type: 'cap' })) || [],
-		stashes: assets.stashes?.map((stash) => curateStash(stash)) || [],
 		fingerprints: assets.fingerprints?.map((fingerprint) => ({
 			hash: fingerprint.hash,
 			type: fingerprint.type,
@@ -143,6 +143,11 @@ function curateScene(rawScene, assets, reqUser, context) {
 		createdBatchId: rawScene.created_batch_id,
 		updatedBatchId: rawScene.updated_batch_id,
 		isNew: assets.lastBatchId === rawScene.created_batch_id,
+		stashes: assets.stashes?.map((stash) => curateStash(stash)) || [],
+		alerts: {
+			only: assets.alerts?.filter((alert) => alert.is_only).flatMap((alert) => alert.alert_ids) || [],
+			multi: assets.alerts?.filter((alert) => !alert.is_only).flatMap((alert) => alert.alert_ids) || [],
+		},
 	};
 
 	const isVideoRestricted = config.media.videoRestrictions.includes(curatedScene.channel.slug) || config.media.videoRestrictions.includes(`_${curatedScene.network?.slug}`);
@@ -179,6 +184,7 @@ export async function fetchScenesById(sceneIds, { reqUser, ...context } = {}) {
 		teasers,
 		fingerprints,
 		stashes,
+		alerts,
 		lastBatch,
 	} = await promiseProps({
 		scenes: knex('releases').whereIn('releases.id', sceneIds),
@@ -356,6 +362,11 @@ export async function fetchScenesById(sceneIds, { reqUser, ...context } = {}) {
 					.where('stashes.user_id', reqUser.id)
 					.whereIn('stashes_scenes.scene_id', sceneIds)
 			: [],
+		alerts: reqUser
+			? knex('alerts_users_scenes')
+					.where('user_id', reqUser.id)
+					.whereIn('scene_id', sceneIds)
+			: [],
 	});
 
 	const actorStashes = reqUser && context.actorStashes
@@ -388,6 +399,7 @@ export async function fetchScenesById(sceneIds, { reqUser, ...context } = {}) {
 		const sceneFingerprints = fingerprints.filter((fingerprint) => fingerprint.scene_id === sceneId);
 		const sceneStashes = stashes.filter((stash) => stash.scene_id === sceneId);
 		const sceneActorStashes = sceneActors.map((actor) => actorStashes.find((stash) => stash.actor_id === actor.id)).filter(Boolean);
+		const sceneAlerts = alerts.filter((alert) => alert.scene_id === sceneId);
 
 		return curateScene(scene, {
 			channel: sceneChannel,
@@ -406,6 +418,7 @@ export async function fetchScenesById(sceneIds, { reqUser, ...context } = {}) {
 			fingerprints: sceneFingerprints,
 			stashes: sceneStashes,
 			actorStashes: sceneActorStashes,
+			alerts: sceneAlerts,
 			lastBatchId: lastBatch?.id,
 		}, reqUser, context);
 	}).filter(Boolean);

src/sync.js

@@ -73,7 +73,7 @@ export async function syncStashes(domain = 'scene', ids) {
 		const docs = searchResponse?.hits?.hits ?? [];
 
 		const orphanedIds = docs
-			.map((hit) => hit.id)
+			.map((hit) => hit._id)
 			.filter((manticoreId) => !validStashedIds.has(manticoreId));
 
 		if (orphanedIds.length === 0) {