fix: reconcile physique targeting for queued posts

This commit is contained in:
jjbrunton committed 2026-08-10 18:23:23 +01:00
1 parent d0a8bd0e2f
commit d8d635e1ee
10 files changed
+250 -12

No files matched your search

@@ -1,5 +1,7 @@
import type Database from 'better-sqlite3';
import type { PhysiqueTargeting } from '../config/index.js';
import type { RedditPost } from '../reddit/api.js';
import { evaluatePhysiqueTargeting } from '../services/physique-targeting.js';
import type { ProcessedPost } from '../services/processor.js';
export type CandidateStatus =
@@ -47,6 +49,11 @@ export interface CandidateEnqueueResult {
deferredAtCapacity: number;
}
export interface PhysiqueTargetingReconciliation {
rejected: number;
retained: number;
}
const ACTIVE_STATUSES: CandidateStatus[] = ['discovered', 'screening', 'ready', 'sending', 'retryable'];
/** Global, durable funnel from Reddit listing discovery through outbound delivery. */
@@ -243,6 +250,54 @@ export class CandidateQueueRepository {
`).run(now, now + sendingDelayMs, now).changes;
}
/** Re-evaluates prepared image candidates that have not yet been leased for sending. */
reconcilePhysiqueTargeting(
targeting: PhysiqueTargeting,
now: number = Date.now(),
): PhysiqueTargetingReconciliation {
const rows = this.db.prepare(`
SELECT post_id, prepared_json FROM candidate_queue
WHERE source_type = 'image'
AND status IN ('ready', 'retryable')
AND prepared_json IS NOT NULL
`).all() as Array<{ post_id: string; prepared_json: string }>;
const preparedRows = rows.map((row) => ({
postId: row.post_id,
prepared: JSON.parse(row.prepared_json) as ProcessedPost,
}));
const reject = this.db.prepare(`
UPDATE candidate_queue
SET status = 'rejected', last_error = ?, updated_at = ?,
lease_owner = NULL, lease_expires_at = NULL, assigned_account_id = NULL
WHERE post_id = ? AND status IN ('ready', 'retryable')
`);
const retain = this.db.prepare(`
UPDATE candidate_queue
SET prepared_json = ?, updated_at = ?
WHERE post_id = ? AND status IN ('ready', 'retryable')
`);
return this.db.transaction(() => {
const result: PhysiqueTargetingReconciliation = { rejected: 0, retained: 0 };
for (const row of preparedRows) {
const score = typeof row.prepared.physiqueScore === 'number'
? row.prepared.physiqueScore
: null;
const decision = evaluatePhysiqueTargeting(score, targeting);
if (!decision.accepted) {
result.rejected += reject.run(decision.reason, now, row.postId).changes;
continue;
}
const refreshed: ProcessedPost = {
...row.prepared,
physiqueTargeting: { ...targeting },
};
result.retained += retain.run(JSON.stringify(refreshed), now, row.postId).changes;
}
return result;
})();
}
expireStale(cutoff: number, now: number = Date.now()): number {
return this.db.prepare(`
UPDATE candidate_queue
+9
View File
@@ -453,11 +453,20 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise<void> {
const workerInstance = `candidate-funnel-${process.pid}-${Date.now()}`;
const leaseMs = 5 * 60_000;
const recovered = candidateQueueRepo.recoverAllLeases();
const physiqueReconciliation = candidateQueueRepo.reconcilePhysiqueTargeting(
config.reddit.physiqueTargeting,
);
app.recipientDeduplicator.recoverStaleReservations(Date.now() - 10 * 60_000);
logger.info(
`Running ${accountRuntimeManager.list().length} account runtime(s) with durable candidate funnel`
+ (recovered > 0 ? `; recovered ${recovered} interrupted candidate(s)` : ''),
);
if (physiqueReconciliation.rejected > 0 || physiqueReconciliation.retained > 0) {
logger.info(
`Reconciled prepared image candidates against physique targeting: `
+ `${physiqueReconciliation.retained} retained, ${physiqueReconciliation.rejected} rejected`,
);
}
await new Promise<void>((resolve) => {
let shuttingDown = false;
const controller = new AbortController();
+56 -3
View File
@@ -106,6 +106,52 @@ function joinText(values: unknown[]): string {
return values.filter((value): value is string => typeof value === 'string').join(' ');
}
function recordValue(value: unknown, key: string): unknown {
return value && typeof value === 'object' ? (value as Record<string, unknown>)[key] : undefined;
}
/** Flatten Reddit's structured profile-link payloads across legacy/current field naming. */
function socialLinkText(userData: unknown): string {
const subreddit = recordValue(userData, 'subreddit');
const candidates = [
recordValue(userData, 'social_links'),
recordValue(userData, 'socialLinks'),
recordValue(subreddit, 'social_links'),
recordValue(subreddit, 'socialLinks'),
];
const values: unknown[] = [];
for (const candidate of candidates) {
if (!Array.isArray(candidate)) continue;
for (const link of candidate) {
values.push(
recordValue(link, 'title'),
recordValue(link, 'handle'),
recordValue(link, 'type'),
recordValue(link, 'url'),
recordValue(link, 'outbound_url'),
recordValue(link, 'outboundUrl'),
);
}
}
return joinText(values);
}
/**
* Catch a narrow paid-content funnel that deliberately keeps the platform name
* out of the bio. Requiring an NSFW profile, an explicit link CTA, and a
* promise of more/exclusive personal content avoids treating ordinary links as
* seller evidence.
*/
function findProfileFunnel(text: string, isNsfw: boolean): string | null {
if (!isNsfw) return null;
const normalized = text.toLocaleLowerCase().replace(/[\u2018\u2019]/g, "'");
const linkCallToAction = /\b(?:hit|check|tap|click|visit|follow)\s+(?:the|my)\s+link\b/;
const contentPromise = /\b(?:see|get|show(?:ing)?|watch)\s+(?:a\s+)?(?:lot\s+)?more(?:\s+of\s+me)?\b|\bexclusive\s+(?:content|pics?|photos?|videos?)\b/;
return linkCallToAction.test(normalized) && contentPromise.test(normalized)
? 'NSFW profile links to more or exclusive personal content'
: null;
}
/** Match configured seller signals in user-controlled text. */
export function findSellerSignal(
text: string,
@@ -422,10 +468,12 @@ export class RedditApiClient {
const userData = response.data?.data;
if (userData) {
const subreddit = userData.subreddit;
const profileText = joinText([
userData.subreddit?.public_description || '',
userData.subreddit?.title || '',
userData.subreddit?.display_name_prefixed || '',
subreddit?.public_description || '',
subreddit?.title || '',
subreddit?.display_name_prefixed || '',
socialLinkText(userData),
]);
const profileSignal = findSellerSignal(profileText, sellerKeywords, sellerDomains);
if (profileSignal) {
@@ -434,6 +482,11 @@ export class RedditApiClient {
);
return true;
}
const funnel = findProfileFunnel(profileText, subreddit?.over_18 === true);
if (funnel) {
this.logger.info(`u/${username} flagged as seller profile (matched funnel: "${funnel}")`);
return true;
}
}
} catch (error) {
const rateLimitError = this.toRateLimitError(error);
+11
View File
@@ -175,6 +175,17 @@ export function createConfigRouter(ctx: WebContext): Router {
configStore.save(ctx.config);
if (parsedPhysiqueTargeting?.success && ctx.services.candidateQueue) {
const reconciliation = ctx.services.candidateQueue.reconcilePhysiqueTargeting(
parsedPhysiqueTargeting.data,
);
ctx.activity.push(
'info',
`Physique targeting reconciled ${reconciliation.retained} prepared image candidate(s)`
+ ` and rejected ${reconciliation.rejected}`,
);
}
ctx.getOnConfigChange()?.(ctx.runtimeConfig);
ctx.activity.push('info', 'Config updated via dashboard');