190 lines
7.4 KiB
TypeScript
190 lines
7.4 KiB
TypeScript
import { RedditRateLimitError, type RedditApiClient, type RedditPost } from '../reddit/api.js';
|
||
import type { PostsRepository, MessagesRepository } from '../database/repositories.js';
|
||
import type { Config } from '../config/index.js';
|
||
import type { Logger } from '../utils/logger.js';
|
||
import { parseGenderTag } from '../utils/gender-tag.js';
|
||
|
||
export class MonitorService {
|
||
// Flips on each call so that when both pools have candidates the bot
|
||
// alternates between image and chat instead of always favouring whichever
|
||
// type the random shuffle put first.
|
||
private chatGoesFirst = false;
|
||
|
||
constructor(
|
||
private redditApi: RedditApiClient,
|
||
private postsRepo: PostsRepository,
|
||
private messagesRepo: MessagesRepository,
|
||
private config: Config,
|
||
private logger: Logger,
|
||
private delayBetweenFetchesMs: number = 1000,
|
||
) { }
|
||
|
||
async getUnprocessedPosts(): Promise<RedditPost[]> {
|
||
const subreddits = this.buildCheckOrder();
|
||
|
||
const targetGenders = this.config.reddit.targetGenders;
|
||
const prioritizeRecent = this.config.reddit.prioritizeRecentPosts;
|
||
const recentCutoffSec = prioritizeRecent
|
||
? Math.floor(Date.now() / 1000) - this.config.reddit.recentWindowMinutes * 60
|
||
: 0;
|
||
// Older posts found while prioritizing recent ones — used as a fallback
|
||
// so the bot still has something to do if no recent posts exist anywhere.
|
||
let fallback: RedditPost | null = null;
|
||
|
||
for (const subreddit of subreddits) {
|
||
try {
|
||
this.logger.debug(`Checking r/${subreddit.name} (${subreddit.type}) for new posts...`);
|
||
const posts = await this.redditApi.getNewPosts(subreddit.name, 25, subreddit.type);
|
||
|
||
// Filter for valid, unprocessed posts
|
||
const validPosts: RedditPost[] = [];
|
||
for (const post of posts) {
|
||
if (this.postsRepo.isProcessed(post.id)) {
|
||
continue;
|
||
}
|
||
|
||
if (this.messagesRepo.hasSentTo(post.author)) {
|
||
this.logger.debug(`Skipping post ${post.id} from u/${post.author} - already messaged this user`);
|
||
continue;
|
||
}
|
||
|
||
if (
|
||
this.config.reddit.filterSellers &&
|
||
(this.config.reddit.sellerKeywords.length > 0 || this.config.reddit.sellerDomains.length > 0)
|
||
) {
|
||
const isSeller = await this.redditApi.isSellerProfile(
|
||
post.author,
|
||
this.config.reddit.sellerKeywords,
|
||
this.config.reddit.sellerDomains,
|
||
post,
|
||
);
|
||
if (isSeller) {
|
||
this.logger.info(`Skipping post ${post.id} from u/${post.author} - seller signals detected`);
|
||
this.postsRepo.insert({
|
||
id: post.id,
|
||
subreddit: post.subreddit,
|
||
author: post.author,
|
||
title: post.title,
|
||
postUrl: this.redditApi.getPostUrl(post),
|
||
imageUrl: subreddit.type === 'image' ? this.redditApi.getImageUrl(post) : null,
|
||
processedAt: Date.now(),
|
||
});
|
||
this.postsRepo.setError(post.id, 'Seller signals detected');
|
||
continue;
|
||
}
|
||
}
|
||
|
||
// Chat subs almost always have a self-declared gender tag in the
|
||
// title/body (e.g. [25F], M4F). Drop posts that don't match the
|
||
// configured targetGenders here so they never enter the processor.
|
||
if (subreddit.type === 'chat' && targetGenders.length > 0) {
|
||
const tag = parseGenderTag(`${post.title}\n${post.selftext ?? ''}`);
|
||
if (!tag) {
|
||
this.logger.debug(`Skipping chat post ${post.id} - no gender tag in title/body`);
|
||
continue;
|
||
}
|
||
if (!targetGenders.includes(tag)) {
|
||
this.logger.debug(
|
||
`Skipping chat post ${post.id} - gender tag "${tag}" not in target list [${targetGenders.join(', ')}]`,
|
||
);
|
||
continue;
|
||
}
|
||
}
|
||
|
||
validPosts.push(post);
|
||
}
|
||
|
||
if (validPosts.length > 0) {
|
||
// Sort by creation time (newest first)
|
||
validPosts.sort((a, b) => b.created_utc - a.created_utc);
|
||
|
||
if (prioritizeRecent) {
|
||
const recent = validPosts.find((p) => p.created_utc >= recentCutoffSec);
|
||
if (recent) {
|
||
this.logger.info(`Found ${validPosts.length} new posts in r/${subreddit.name} (recent match)`);
|
||
return [recent];
|
||
}
|
||
// Remember the freshest non-recent candidate as a fallback and
|
||
// keep scanning other subs for an actually-recent post.
|
||
if (!fallback || validPosts[0].created_utc > fallback.created_utc) {
|
||
fallback = validPosts[0];
|
||
}
|
||
} else {
|
||
this.logger.info(`Found ${validPosts.length} new posts in r/${subreddit.name}`);
|
||
// Return just the first (newest) valid post
|
||
// This ensures we switch to a different random subreddit on the next poll cycle
|
||
// instead of processing a batch from the same subreddit
|
||
return [validPosts[0]];
|
||
}
|
||
}
|
||
} catch (error) {
|
||
if (error instanceof RedditRateLimitError) throw error;
|
||
|
||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||
this.logger.error(`Failed to fetch posts from r/${subreddit.name}: ${errorMessage}`);
|
||
}
|
||
|
||
// Small delay between subreddit fetches to be nice to Reddit
|
||
if (this.delayBetweenFetchesMs > 0) {
|
||
await this.sleep(this.delayBetweenFetchesMs);
|
||
}
|
||
}
|
||
|
||
if (fallback) {
|
||
this.logger.info(
|
||
`No posts within last ${this.config.reddit.recentWindowMinutes}m — falling back to newest available (r/${fallback.subreddit})`,
|
||
);
|
||
return [fallback];
|
||
}
|
||
|
||
return [];
|
||
}
|
||
|
||
/**
|
||
* Build an interleaved check order that balances image and chat subs.
|
||
*
|
||
* Each pool is shuffled independently, then zipped together so a chat sub
|
||
* is checked roughly every other position even when there are far more
|
||
* image subs. `chatGoesFirst` flips per call so neither type is
|
||
* permanently advantaged when both pools have fresh posts.
|
||
*/
|
||
private buildCheckOrder(): Config['reddit']['subreddits'] {
|
||
const images = this.weightedShuffle(
|
||
this.config.reddit.subreddits.filter((s) => s.type === 'image'),
|
||
);
|
||
const chats = this.weightedShuffle(
|
||
this.config.reddit.subreddits.filter((s) => s.type === 'chat'),
|
||
);
|
||
|
||
const [a, b] = this.chatGoesFirst ? [chats, images] : [images, chats];
|
||
this.chatGoesFirst = !this.chatGoesFirst;
|
||
|
||
const out: Config['reddit']['subreddits'] = [];
|
||
const max = Math.max(a.length, b.length);
|
||
for (let i = 0; i < max; i++) {
|
||
if (i < a.length) out.push(a[i]);
|
||
if (i < b.length) out.push(b[i]);
|
||
}
|
||
return out;
|
||
}
|
||
|
||
// Weighted random ordering (Efraimidis–Spirakis): key = random^(1/weight),
|
||
// sorted descending. Heavier-weighted subs are more likely to land earlier
|
||
// each cycle, while every sub still has a non-zero chance of any position.
|
||
private weightedShuffle<T extends { weight?: number }>(items: T[]): T[] {
|
||
return items
|
||
.map((item) => {
|
||
const w = item.weight && item.weight > 0 ? item.weight : 1;
|
||
const r = Math.random();
|
||
const key = r === 0 ? -Infinity : Math.log(r) / w;
|
||
return { item, key };
|
||
})
|
||
.sort((a, b) => b.key - a.key)
|
||
.map(({ item }) => item);
|
||
}
|
||
|
||
private sleep(ms: number): Promise<void> {
|
||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||
}
|
||
}
|