diff --git a/src/alerts.js b/src/alerts.js
index a4d6e8e..a8bab88 100755
--- a/src/alerts.js
+++ b/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)) {
- throw new HttpError('Alert must contain at least one actor, tag or entity', 400);
+ 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)) {
@@ -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'),
- alert.tags?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_tags'),
- alert.entities?.length > 0 && knex.schema.refreshMaterializedView('alerts_users_entities'),
+ 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]);
@@ -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,
+ });
+}
diff --git a/src/app.js b/src/app.js
index 0030683..7e5672e 100644
--- a/src/app.js
+++ b/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();
diff --git a/src/scenes.js b/src/scenes.js
index 821f166..51aac0c 100644
--- a/src/scenes.js
+++ b/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);
diff --git a/src/sync.js b/src/sync.js
index 6c72ab0..8bf7ebe 100644
--- a/src/sync.js
+++ b/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) {