703 lines
30 KiB
TypeScript
703 lines
30 KiB
TypeScript
import crypto from 'node:crypto';
|
|
import path from 'node:path';
|
|
import type Database from 'better-sqlite3';
|
|
import type { Config } from '../config/index.js';
|
|
import type { Logger } from '../utils/logger.js';
|
|
import { BrowserClient } from '../browser/client.js';
|
|
import {
|
|
PostsRepository, MessagesRepository, SuccessfulMessagesRepository,
|
|
DeliveryOutboxRepository, ConversationStateRepository,
|
|
ConversationReadStateRepository,
|
|
ConversationAutomationClaimRepository,
|
|
ConversationOutcomesRepository, RedditUsersRepository,
|
|
ActivityLogRepository,
|
|
RecipientHistoryRepository,
|
|
} from '../database/repositories.js';
|
|
import {
|
|
AccountContentRepository,
|
|
RedditAccountEstablishmentRepository,
|
|
RedditAccountsRepository,
|
|
RedditRecipientAccountStatusRepository,
|
|
} from '../database/account-repositories.js';
|
|
import { RedditApiClient, RedditRateLimitError, type RedditPost } from '../reddit/api.js';
|
|
import {
|
|
REDDIT_WRITE_ORIGIN,
|
|
RedditCommentClient,
|
|
createAnonymousFetcher,
|
|
createBrowserWriteTransport,
|
|
} from '../reddit/comment-client.js';
|
|
import { RedditRequestCoordinator } from '../reddit/request-coordinator.js';
|
|
import { OpenRouterClient } from '../ai/openrouter.js';
|
|
import { MonitorService } from './monitor.js';
|
|
import { ProcessorService, isSkip, type ProcessResult, type ProcessedPost } from './processor.js';
|
|
import { ResponseCheckerService } from './response-checker.js';
|
|
import { StrategySelector } from './strategy-selector.js';
|
|
import {
|
|
AccountAuthContext,
|
|
normalizeRedditAccountId,
|
|
type AuthenticatedAccount,
|
|
type RedditAccountIdentity,
|
|
} from './account-auth-context.js';
|
|
import { ChatCache } from './chat/cache.js';
|
|
import {
|
|
BrowserMatrixTokenRefresher, ChatService, MatrixAuthError, MatrixClient,
|
|
MatrixMediaDownloader, MatrixUsernameResolver,
|
|
type MatrixCredentials, type UsernameResolver,
|
|
} from './chat/index.js';
|
|
import { MediaService } from './media.js';
|
|
import { MediaSyncService } from './media-sync.js';
|
|
import { RateLimiter } from '../utils/rate-limiter.js';
|
|
import {
|
|
RedditAccountRuntime, RedditAccountRuntimeFactory,
|
|
accountBrowserProfileDir, type RedditAccountRuntimeResources,
|
|
type VerifiedAccountSession, type PreparedDeliveryResult,
|
|
} from './reddit-account-runtime.js';
|
|
import { RecipientDeduplicator } from './recipient-deduplicator.js';
|
|
import { ConfigStore } from '../config/persistence.js';
|
|
import { createConfigBackedSubredditPollingControl } from './subreddit-polling-control.js';
|
|
import { ConversationAutomationService } from './conversation-automation.js';
|
|
import { AiUsageRepository } from '../database/ai-usage-repository.js';
|
|
import { MediaLibraryRepository } from '../database/media-library-repository.js';
|
|
import { MediaLibraryExchangeService } from './media-library-exchange.js';
|
|
import { RecipientAccountStatusService } from './recipient-account-status.js';
|
|
import { AccountEstablishmentService } from './account-establishment.js';
|
|
import { AccountGrowthService, type GrowthActivityEvent } from './account-growth.js';
|
|
import { AccountGrowthRepository } from '../database/growth-repository.js';
|
|
import { AccountBrowsingSimulationRepository } from '../database/browsing-simulation-repository.js';
|
|
import { AccountBrowsingSimulationService } from './account-browsing-simulation.js';
|
|
import { allowsConversationSync, allowsSending, type BotMode } from './bot-mode.js';
|
|
|
|
export interface ConcreteRuntimeFactoryOptions {
|
|
config: Config;
|
|
logger: Logger;
|
|
db: Database.Database;
|
|
getBotMode?: () => BotMode;
|
|
}
|
|
|
|
export function materializePreparedPostForAccount(
|
|
resources: Pick<RedditAccountRuntimeResources, 'postsRepo' | 'redditApi'>,
|
|
processed: ProcessedPost,
|
|
now: number = Date.now(),
|
|
): void {
|
|
const { post } = processed;
|
|
resources.postsRepo.insertIfAbsent({
|
|
id: post.id,
|
|
subreddit: post.subreddit,
|
|
author: post.author,
|
|
title: post.title,
|
|
postUrl: resources.redditApi.getPostUrl(post),
|
|
imageUrl: processed.sourceType === 'image' ? resources.redditApi.getImageUrl(post) : null,
|
|
postCreatedAt: post.created_utc * 1_000,
|
|
processedAt: now,
|
|
});
|
|
resources.postsRepo.updateAnalysis(post.id, processed.imageAnalysis, processed.generatedMessage);
|
|
}
|
|
|
|
export function createConcreteRedditAccountRuntimeFactory(
|
|
options: ConcreteRuntimeFactoryOptions,
|
|
): RedditAccountRuntimeFactory {
|
|
const recipientDeduplicator = new RecipientDeduplicator(
|
|
new RecipientHistoryRepository(options.db),
|
|
);
|
|
const redditRequestCoordinator = new RedditRequestCoordinator();
|
|
return new RedditAccountRuntimeFactory(
|
|
(session) => buildRuntime(options, session, recipientDeduplicator, redditRequestCoordinator),
|
|
);
|
|
}
|
|
|
|
async function buildRuntime(
|
|
{ config, logger: rootLogger, db, getBotMode = () => 'running' }: ConcreteRuntimeFactoryOptions,
|
|
session: VerifiedAccountSession,
|
|
recipientDeduplicator: RecipientDeduplicator,
|
|
redditRequestCoordinator: RedditRequestCoordinator,
|
|
): Promise<RedditAccountRuntime> {
|
|
const logger = rootLogger.child({
|
|
accountId: session.accountId,
|
|
redditUsername: session.redditUsername,
|
|
});
|
|
const browserProfileDir = accountBrowserProfileDir(
|
|
path.resolve(process.cwd(), config.browser.sessionDir),
|
|
session.accountId,
|
|
);
|
|
const accountConfig: Config = {
|
|
...config,
|
|
browser: { ...config.browser, sessionDir: browserProfileDir },
|
|
web: { ...config.web, enabled: false },
|
|
};
|
|
|
|
const browserClient = new BrowserClient(accountConfig, logger);
|
|
await browserClient.initialize();
|
|
try {
|
|
await browserClient.importRedditCookies(session.cookies);
|
|
const identity = await browserClient.getAuthenticatedIdentity();
|
|
const expectedRedditId = matrixRedditId(session.matrixUserId);
|
|
if (!identity || !expectedRedditId || identity.accountId.toLowerCase().replace(/^t2_/, '') !== expectedRedditId) {
|
|
throw new Error(`Stored Reddit cookies no longer verify for u/${session.redditUsername}`);
|
|
}
|
|
|
|
const matrixClient = new MatrixClient(session.matrixCredentials, logger);
|
|
const accountsRepo = new RedditAccountsRepository(db);
|
|
const tokenRefresher = new BrowserMatrixTokenRefresher({
|
|
context: browserClient.getContext()!,
|
|
expectedUserId: session.matrixUserId,
|
|
verifyCredentials: (credentials) => new MatrixClient(credentials, logger).whoami(),
|
|
applyCredentials: (credentials) => matrixClient.updateCredentials(credentials),
|
|
persistCredentials: (credentials: MatrixCredentials) => {
|
|
session.matrixCredentials = credentials;
|
|
accountsRepo.updateMatrixCredentials(session.accountId, JSON.stringify(credentials));
|
|
},
|
|
logger,
|
|
});
|
|
const verifyMatrixSession = async (refreshOnAuthError: boolean): Promise<boolean> => {
|
|
try {
|
|
return await matrixClient.whoami() === session.matrixUserId;
|
|
} catch (error) {
|
|
if (!refreshOnAuthError || !(error instanceof MatrixAuthError)) throw error;
|
|
logger.warn(`Stored Matrix token for u/${session.redditUsername} was rejected — refreshing from the browser session`);
|
|
await tokenRefresher.refresh();
|
|
return await matrixClient.whoami() === session.matrixUserId;
|
|
}
|
|
};
|
|
if (!(await verifyMatrixSession(true))) {
|
|
throw new Error(`Stored Matrix session no longer verifies for u/${session.redditUsername}`);
|
|
}
|
|
|
|
const accountAuth = new AccountAuthContext(logger);
|
|
accountAuth.markAuthenticated(runtimeAuthenticatedAccount(session, identity));
|
|
accountAuth.setAuthenticator(async () => {
|
|
const current = await browserClient.getAuthenticatedIdentity();
|
|
if (!current || matrixRedditId(session.matrixUserId) !== current.accountId.toLowerCase().replace(/^t2_/, '')) return null;
|
|
return await verifyMatrixSession(true)
|
|
? runtimeAuthenticatedAccount(session, current)
|
|
: null;
|
|
});
|
|
|
|
const postsRepo = new PostsRepository(db, session.accountId);
|
|
const messagesRepo = new MessagesRepository(db, session.accountId);
|
|
const successfulMessagesRepo = new SuccessfulMessagesRepository(db, session.accountId);
|
|
const deliveryOutboxRepo = new DeliveryOutboxRepository(db, session.accountId);
|
|
const conversationStateRepo = new ConversationStateRepository(db, session.accountId);
|
|
const conversationReadStateRepo = new ConversationReadStateRepository(db, session.accountId);
|
|
const conversationAutomationClaimRepo = new ConversationAutomationClaimRepository(db, session.accountId);
|
|
const conversationOutcomesRepo = new ConversationOutcomesRepository(db, session.accountId);
|
|
const redditUsersRepo = new RedditUsersRepository(db);
|
|
const accountContentRepo = new AccountContentRepository(db);
|
|
const activityLogRepo = new ActivityLogRepository(db, session.accountId);
|
|
|
|
const redditApi = new RedditApiClient(accountConfig, logger, redditRequestCoordinator);
|
|
redditApi.setBrowserContext(browserClient.getContext()!);
|
|
const chatPage = await browserClient.getContext()!.newPage();
|
|
const aiUsageRepo = new AiUsageRepository(db);
|
|
const aiClient = new OpenRouterClient(accountConfig, logger, undefined, (usage) => {
|
|
aiUsageRepo.record({ ...usage, accountId: session.accountId });
|
|
});
|
|
const monitorService = new MonitorService(
|
|
redditApi, postsRepo, recipientDeduplicator, accountConfig, logger,
|
|
1000, createConfigBackedSubredditPollingControl(config, new ConfigStore(db), logger),
|
|
);
|
|
const processorService = new ProcessorService(
|
|
redditApi,
|
|
aiClient,
|
|
accountConfig,
|
|
logger,
|
|
undefined,
|
|
(sourceType) => messagesRepo
|
|
.getRecent(20)
|
|
.filter((message) => message.sourceType === sourceType || (sourceType === 'image' && !message.sourceType))
|
|
.map((message) => message.message),
|
|
);
|
|
processorService.setStrategySelector(new StrategySelector(accountConfig, messagesRepo, logger));
|
|
processorService.setChatStrategySelector(new StrategySelector(accountConfig, messagesRepo, logger, 'chatStrategies'));
|
|
|
|
const mediaDownloader = new MatrixMediaDownloader(matrixClient, logger);
|
|
const mediaService = new MediaService(
|
|
mediaDownloader,
|
|
logger,
|
|
undefined,
|
|
accountContentRepo,
|
|
() => session.accountId,
|
|
);
|
|
const usernameResolverImpl = new MatrixUsernameResolver(matrixClient);
|
|
const usernameResolver: UsernameResolver = {
|
|
resolveT2Ids: (ids) => usernameResolverImpl.resolveT2Ids(ids),
|
|
resolveUsernameToT2: (username) => redditApi.resolveUsernameToT2(username),
|
|
};
|
|
const chatCache = new ChatCache(logger, undefined, accountContentRepo, session.accountId);
|
|
const chatService = new ChatService(
|
|
matrixClient, logger, successfulMessagesRepo, mediaService,
|
|
tokenRefresher, usernameResolver, chatCache,
|
|
messagesRepo, redditUsersRepo, accountAuth, conversationOutcomesRepo, conversationStateRepo,
|
|
);
|
|
await chatService.loadCache();
|
|
const mediaSyncService = new MediaSyncService(chatService, logger, accountConfig.media.syncIntervalMs, accountAuth);
|
|
const recipientAccountStatusRepo = new RedditRecipientAccountStatusRepository(db);
|
|
const recipientAccountStatusService = new RecipientAccountStatusService(
|
|
chatService,
|
|
redditApi,
|
|
recipientAccountStatusRepo,
|
|
logger,
|
|
accountAuth,
|
|
);
|
|
const establishmentService = new AccountEstablishmentService(
|
|
session.accountId,
|
|
session.redditUsername,
|
|
redditApi,
|
|
new RedditAccountEstablishmentRepository(db),
|
|
accountConfig.accountEstablishment,
|
|
logger,
|
|
accountAuth,
|
|
);
|
|
const responseChecker = new ResponseCheckerService(
|
|
chatPage, messagesRepo, successfulMessagesRepo, postsRepo, logger,
|
|
accountAuth, conversationOutcomesRepo,
|
|
);
|
|
// Comment posting runs through the account's own browser page so Reddit sees
|
|
// the same session as a human using this profile, and shares the process-wide
|
|
// request pacer so writes never burst alongside listing reads.
|
|
const growthPage = await browserClient.getContext()!.newPage();
|
|
const commentClient = new RedditCommentClient(
|
|
createBrowserWriteTransport(growthPage),
|
|
logger,
|
|
{
|
|
scheduler: redditRequestCoordinator,
|
|
anonymousFetch: createAnonymousFetcher(accountConfig.reddit.userAgent),
|
|
// Names and domains only — never cookie values. This is the fastest way
|
|
// to see why Reddit stopped recognising a session on the write origin.
|
|
describeCookies: async () => {
|
|
const cookies = await browserClient.getContext()?.cookies(REDDIT_WRITE_ORIGIN) ?? [];
|
|
if (cookies.length === 0) return `none visible to ${REDDIT_WRITE_ORIGIN}`;
|
|
return cookies.map((cookie) => `${cookie.name}@${cookie.domain}`).join(', ');
|
|
},
|
|
},
|
|
);
|
|
const growthRepository = new AccountGrowthRepository(db);
|
|
const growthService = new AccountGrowthService({
|
|
accountId: session.accountId,
|
|
redditUsername: session.redditUsername,
|
|
repository: growthRepository,
|
|
reddit: {
|
|
getHotPosts: (subreddit, limit) => redditApi.getHotPosts(subreddit, limit),
|
|
getThreadTopComments: (subreddit, postId, limit) =>
|
|
redditApi.getThreadTopComments(subreddit, postId, limit),
|
|
},
|
|
commentClient,
|
|
ai: aiClient,
|
|
auth: accountAuth,
|
|
logger,
|
|
dryRun: accountConfig.browser.dryRun,
|
|
onActivity: (event) => recordGrowthActivity(activityLogRepo, session.redditUsername, event),
|
|
});
|
|
const browsingRepository = new AccountBrowsingSimulationRepository(db);
|
|
const browsingService = new AccountBrowsingSimulationService({
|
|
accountId: session.accountId,
|
|
repository: browsingRepository,
|
|
growthSubreddits: (now) => growthRepository.listActiveSubreddits(session.accountId, now),
|
|
reddit: {
|
|
getHotPosts: (subreddit, limit) => redditApi.getHotPosts(subreddit, limit),
|
|
getBrowsingComments: (subreddit, postId, limit) =>
|
|
redditApi.getBrowsingComments(subreddit, postId, limit),
|
|
},
|
|
sink: browsingRepository,
|
|
logger,
|
|
onActivity: (decision) => activityLogRepo.insert({
|
|
timestamp: decision.decidedAt,
|
|
type: 'response',
|
|
message: `Simulated ${decision.thingKind} upvote in r/${decision.subreddit}; no Reddit vote was submitted`,
|
|
postUrl: `https://www.reddit.com${decision.permalink}`,
|
|
postTitle: `r/${decision.subreddit}`,
|
|
}),
|
|
});
|
|
const rateLimiter = new RateLimiter(accountConfig, logger, undefined, {
|
|
sentMessageHistory: async (since) => messagesRepo.getSentTimestampsSince(since),
|
|
});
|
|
await rateLimiter.initialize();
|
|
const mediaExchange = new MediaLibraryExchangeService(
|
|
new MediaLibraryRepository(db),
|
|
chatService,
|
|
logger,
|
|
session.accountId,
|
|
undefined,
|
|
accountConfig.browser.dryRun,
|
|
);
|
|
const conversationAutomation = new ConversationAutomationService(
|
|
chatService,
|
|
aiClient,
|
|
conversationStateRepo,
|
|
messagesRepo,
|
|
conversationAutomationClaimRepo,
|
|
rateLimiter,
|
|
logger,
|
|
accountConfig.browser.dryRun,
|
|
accountConfig.reddit.filterSellers ? accountConfig.reddit.sellerKeywords : [],
|
|
accountConfig.reddit.filterSellers ? accountConfig.reddit.sellerDomains : [],
|
|
() => accountConfig.reddit.subreddits,
|
|
Date.now,
|
|
undefined,
|
|
undefined,
|
|
mediaExchange,
|
|
// Read through to the live config object rather than snapshotting, so a
|
|
// dashboard configuration save applies to a running account without a
|
|
// restart. The service reads these on every inbound reply.
|
|
{
|
|
get offPlatformKill() { return accountConfig.conversations.offPlatformKill; },
|
|
get killOnSellerKeyword() { return accountConfig.conversations.killOnSellerKeyword; },
|
|
triage: conversationReadStateRepo,
|
|
screener: aiClient,
|
|
get screening() { return accountConfig.conversations.aiReplyScreening; },
|
|
},
|
|
);
|
|
|
|
const resources: RedditAccountRuntimeResources = {
|
|
browserProfileDir, browserClient, accountAuth, matrixClient, chatCache, redditApi,
|
|
postsRepo, messagesRepo, successfulMessagesRepo, deliveryOutboxRepo,
|
|
conversationStateRepo, conversationReadStateRepo, conversationOutcomesRepo,
|
|
redditUsersRepo, accountContentRepo, recipientAccountStatusRepo,
|
|
activityLogRepo,
|
|
monitorService, processorService, responseChecker, conversationAutomation, chatService, mediaService,
|
|
mediaSyncService, mediaExchange, rateLimiter,
|
|
recipientAccountStatusService, establishmentService, growthService, browsingService,
|
|
};
|
|
const lifecycle = new AccountRuntimeLoop(
|
|
resources, accountConfig, logger, getBotMode, recipientDeduplicator,
|
|
);
|
|
return new RedditAccountRuntime(session, resources, lifecycle);
|
|
} catch (error) {
|
|
await browserClient.close();
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
/** Keeps the Reddit auth fullname separate from the Matrix-shaped runtime/database key. */
|
|
export function runtimeAuthenticatedAccount(
|
|
session: Pick<VerifiedAccountSession, 'matrixUserId'>,
|
|
identity: RedditAccountIdentity,
|
|
): AuthenticatedAccount {
|
|
return {
|
|
accountId: normalizeRedditAccountId(identity.accountId),
|
|
redditUsername: identity.username,
|
|
matrixUserId: session.matrixUserId,
|
|
};
|
|
}
|
|
|
|
/** Exported for tests; production code reaches it through the runtime factory. */
|
|
export class AccountRuntimeLoop {
|
|
private controller: AbortController | null = null;
|
|
private loop: Promise<void> | null = null;
|
|
private mediaSyncRunning = false;
|
|
private startupSyncStarted = false;
|
|
|
|
constructor(
|
|
private readonly resources: RedditAccountRuntimeResources,
|
|
private readonly config: Config,
|
|
private readonly logger: Logger,
|
|
private readonly getBotMode: () => BotMode,
|
|
private readonly recipientDeduplicator: RecipientDeduplicator,
|
|
) {}
|
|
|
|
start(): void {
|
|
if (this.loop) return;
|
|
this.controller = new AbortController();
|
|
this.resources.recipientAccountStatusService.start();
|
|
this.resources.establishmentService.start();
|
|
// A comment interrupted mid-post may already be live on Reddit, so those
|
|
// rows are retired before any new growth work begins.
|
|
this.resources.growthService?.recoverAfterRestart();
|
|
this.loop = this.run(this.controller.signal).finally(() => { this.loop = null; });
|
|
}
|
|
|
|
async stop(): Promise<void> {
|
|
this.controller?.abort();
|
|
this.mediaSyncRunning = false;
|
|
this.resources.mediaSyncService.stop();
|
|
this.resources.recipientAccountStatusService.stop();
|
|
this.resources.establishmentService.stop();
|
|
await this.loop?.catch((error) => {
|
|
if (!(error instanceof DOMException && error.name === 'AbortError')) throw error;
|
|
});
|
|
await this.resources.browserClient.close();
|
|
}
|
|
|
|
/**
|
|
* Starts or stops the background media sync only on a mode transition —
|
|
* `MediaSyncService.start()` re-wraps its own `stop()` on every call, so
|
|
* calling it once per loop tick would pile up closures.
|
|
*/
|
|
private reconcileMediaSync(enabled: boolean): void {
|
|
if (enabled === this.mediaSyncRunning) return;
|
|
this.mediaSyncRunning = enabled;
|
|
if (!enabled) {
|
|
this.resources.mediaSyncService.stop();
|
|
return;
|
|
}
|
|
this.resources.mediaSyncService.start();
|
|
if (this.startupSyncStarted) return;
|
|
this.startupSyncStarted = true;
|
|
void this.resources.chatService.runStartupSync()
|
|
.then(() => this.resources.chatService.enrichUsernames())
|
|
.catch((error) => this.logger.error(`[account ${this.resources.accountAuth.getSnapshot().redditUsername}] startup sync failed:`, error));
|
|
}
|
|
|
|
private async run(signal: AbortSignal): Promise<void> {
|
|
let nextReplyCheckAt = 0;
|
|
let nextPerformanceCheckAt = 0;
|
|
let nextGrowthCheckAt = 0;
|
|
let nextBrowsingCheckAt = 0;
|
|
const replyInterval = 30_000;
|
|
const performanceInterval = Math.max(this.config.reddit.pollIntervalMs * 5, 60_000);
|
|
// The pacer decides whether a comment is actually due; this is just how
|
|
// often the account asks the question.
|
|
const growthInterval = 60_000;
|
|
while (!signal.aborted) {
|
|
try {
|
|
const mode = this.getBotMode();
|
|
// Background media sync follows the mode too, so `stopped` really does
|
|
// halt every background refresh rather than leaving a 30-minute one.
|
|
this.reconcileMediaSync(allowsConversationSync(mode));
|
|
if (!allowsConversationSync(mode)) {
|
|
await delay(1_000, signal);
|
|
continue;
|
|
}
|
|
const auth = this.resources.accountAuth;
|
|
if (!auth.isFullyAuthenticated() && !(await auth.ensureFullyAuthenticated())) {
|
|
await delay(1_000, signal);
|
|
continue;
|
|
}
|
|
if (Date.now() >= nextReplyCheckAt) {
|
|
nextReplyCheckAt = Date.now() + replyInterval;
|
|
// In scan mode the same pass refreshes chats and counts actionable
|
|
// replies, but stops short of generating or sending any of them.
|
|
const automationResults = await this.resources.conversationAutomation.processPendingConversations(
|
|
signal,
|
|
{ scanOnly: !allowsSending(mode) },
|
|
);
|
|
for (const result of automationResults) {
|
|
if (result.kind === 'sent') continue;
|
|
this.resources.activityLogRepo.insert({
|
|
timestamp: Date.now(),
|
|
type: 'response',
|
|
message: result.kind === 'human-required'
|
|
? `Human follow-up required for u/${result.username}`
|
|
: result.kind === 'image-received'
|
|
? `Image received from u/${result.username}`
|
|
: `Automation stopped for u/${result.username} after a boundary`,
|
|
recipient: result.username,
|
|
});
|
|
}
|
|
}
|
|
if (Date.now() >= nextPerformanceCheckAt) {
|
|
nextPerformanceCheckAt = Date.now() + performanceInterval;
|
|
await this.resources.responseChecker.checkForResponses();
|
|
}
|
|
// Growth is the lowest-priority work this account does, so it runs after
|
|
// replies have been handled in this same pass, and it is isolated so a
|
|
// growth failure can never stop reply handling or response checking.
|
|
// It deliberately still runs when the account is held out of outbound
|
|
// messaging: building a comment history is what lifts that hold.
|
|
if (this.resources.growthService && Date.now() >= nextGrowthCheckAt) {
|
|
nextGrowthCheckAt = Date.now() + growthInterval;
|
|
try {
|
|
await this.resources.growthService.tick({ allowPosting: allowsSending(mode) }, signal);
|
|
} catch (error) {
|
|
if (signal.aborted) return;
|
|
this.logger.error('Account growth tick failed:', error);
|
|
}
|
|
}
|
|
// Browsing simulation is strictly below comment growth. It records a
|
|
// hypothetical decision only and runs in both scan and running modes.
|
|
if (this.resources.browsingService && Date.now() >= nextBrowsingCheckAt) {
|
|
nextBrowsingCheckAt = Date.now() + growthInterval;
|
|
try {
|
|
await this.resources.browsingService.tick(signal);
|
|
} catch (error) {
|
|
if (signal.aborted) return;
|
|
this.logger.error('Browsing simulation tick failed:', error);
|
|
}
|
|
}
|
|
await delay(5_000, signal);
|
|
} catch (error) {
|
|
if (signal.aborted) return;
|
|
if (error instanceof RedditRateLimitError) {
|
|
await delay(error.retryAfterMs + 1_000, signal);
|
|
continue;
|
|
}
|
|
this.logger.error(`Account runtime loop failed for ${this.resources.accountAuth.getSnapshot().redditUsername}:`, error);
|
|
await delay(60_000, signal);
|
|
}
|
|
}
|
|
}
|
|
|
|
async dispatchPost(post: RedditPost, signal?: AbortSignal): Promise<void> {
|
|
const processed = await this.preparePost(post, signal);
|
|
if (isSkip(processed)) return;
|
|
await this.dispatchPreparedPost(processed, signal);
|
|
}
|
|
|
|
async preparePost(post: RedditPost, signal?: AbortSignal): Promise<ProcessResult> {
|
|
const { postsRepo, redditApi, processorService, responseChecker } = this.resources;
|
|
if (signal?.aborted) throw new DOMException('Operation aborted', 'AbortError');
|
|
const subreddit = this.config.reddit.subreddits.find((entry) => entry.name.toLowerCase() === post.subreddit.toLowerCase());
|
|
const isChat = subreddit?.type === 'chat';
|
|
postsRepo.insertIfAbsent({
|
|
id: post.id, subreddit: post.subreddit, author: post.author, title: post.title,
|
|
postUrl: redditApi.getPostUrl(post), imageUrl: isChat ? null : redditApi.getImageUrl(post),
|
|
postCreatedAt: post.created_utc * 1_000, processedAt: Date.now(),
|
|
});
|
|
processorService.setSuccessfulExamples(responseChecker.getSuccessfulExamples(3, post.subreddit));
|
|
const processed = await processorService.processPost(post);
|
|
if (isSkip(processed)) {
|
|
postsRepo.setError(post.id, processed.skipReason);
|
|
this.resources.activityLogRepo.insert({
|
|
timestamp: Date.now(), type: 'skipped',
|
|
message: `Skipped post by u/${post.author} (${processed.skipReason})`, recipient: post.author,
|
|
});
|
|
return processed;
|
|
}
|
|
postsRepo.updateAnalysis(post.id, processed.imageAnalysis, processed.generatedMessage);
|
|
return processed;
|
|
}
|
|
|
|
async dispatchPreparedPost(processed: ProcessedPost, signal?: AbortSignal): Promise<PreparedDeliveryResult> {
|
|
const { postsRepo, messagesRepo, redditApi, rateLimiter, chatService, establishmentService } = this.resources;
|
|
const { post } = processed;
|
|
if (signal?.aborted) {
|
|
return { status: 'retryable', reason: 'Delivery interrupted by shutdown', retryAt: Date.now() + 5_000 };
|
|
}
|
|
|
|
// Preparation may have been performed by a different authenticated
|
|
// scanner. Sent messages are account-scoped and reference that account's
|
|
// processed-post row, so materialise the prepared candidate locally before
|
|
// performing the irreversible remote chat creation.
|
|
materializePreparedPostForAccount(this.resources, processed);
|
|
|
|
return rateLimiter.throttle(async (): Promise<PreparedDeliveryResult> => {
|
|
if (this.config.browser.dryRun) {
|
|
const reason = 'Dry run: message prepared but not sent';
|
|
postsRepo.setError(post.id, reason);
|
|
return { status: 'terminal', reason };
|
|
}
|
|
if (!this.recipientDeduplicator.tryReserve(post.author)) {
|
|
const reason = 'User already contacted across accounts';
|
|
postsRepo.setError(post.id, reason);
|
|
return { status: 'terminal', reason };
|
|
}
|
|
|
|
const start = await chatService.startConversationDetailed(
|
|
post.author,
|
|
processed.generatedMessage,
|
|
post.author_fullname,
|
|
);
|
|
if (start.status !== 'sent') {
|
|
this.recipientDeduplicator.release(post.author);
|
|
// The rejection reason is the only place Reddit tells us the account is
|
|
// too new to invite anyone; the code, when there is one, rides on the
|
|
// cooldown the chat service just recorded.
|
|
establishmentService.recordSendFailure(
|
|
start.reason,
|
|
chatService.getInvitationCooldown()?.redditCode ?? null,
|
|
);
|
|
if (start.status === 'terminal') {
|
|
postsRepo.setError(post.id, start.reason);
|
|
this.resources.activityLogRepo.insert({
|
|
timestamp: Date.now(), type: 'skipped',
|
|
message: `Skipped u/${post.author} (${start.reason})`, recipient: post.author,
|
|
});
|
|
}
|
|
return start;
|
|
}
|
|
|
|
postsRepo.markMessageSent(post.id);
|
|
establishmentService.recordSendSuccess();
|
|
messagesRepo.insert({
|
|
postId: post.id, recipient: post.author, subject: post.author,
|
|
message: processed.generatedMessage, sentAt: Date.now(),
|
|
strategyId: processed.strategyId, chatRoomId: start.chat.roomId,
|
|
experimentId: processed.experimentId, promptVersion: processed.promptVersion,
|
|
modelId: processed.modelId, selectionReason: processed.selectionReason,
|
|
sourceType: processed.sourceType,
|
|
});
|
|
this.recipientDeduplicator.commit(post.author);
|
|
this.resources.activityLogRepo.insert({
|
|
timestamp: Date.now(), type: 'sent', message: `Sent message to u/${post.author} (r/${post.subreddit})`,
|
|
recipient: post.author,
|
|
imageUrl: processed.sourceType === 'image' ? redditApi.getImageUrl(post) ?? undefined : undefined,
|
|
postTitle: post.title, postUrl: redditApi.getPostUrl(post),
|
|
});
|
|
void chatService.refreshTopChats(5);
|
|
return { status: 'sent', conversationId: start.chat.roomId! };
|
|
}, (result) => result.status === 'sent', signal);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Mirrors growth events into the operator activity log. Growth posts publicly
|
|
* without a human in the loop, so every outcome has to leave a trail.
|
|
*/
|
|
function recordGrowthActivity(
|
|
activityLogRepo: ActivityLogRepository,
|
|
redditUsername: string,
|
|
event: GrowthActivityEvent,
|
|
): void {
|
|
const timestamp = Date.now();
|
|
switch (event.kind) {
|
|
case 'posted':
|
|
activityLogRepo.insert({
|
|
timestamp,
|
|
type: 'sent',
|
|
message: `Growth comment posted to r/${event.subreddit} as u/${redditUsername}: ${event.comment}`,
|
|
...(event.permalink
|
|
? { postUrl: `https://www.reddit.com${event.permalink}`, postTitle: `r/${event.subreddit}` }
|
|
: {}),
|
|
});
|
|
return;
|
|
case 'rejected':
|
|
activityLogRepo.insert({
|
|
timestamp,
|
|
type: 'skipped',
|
|
message: `Growth comment for r/${event.subreddit} discarded (${event.reason})`,
|
|
});
|
|
return;
|
|
case 'removed':
|
|
activityLogRepo.insert({
|
|
timestamp,
|
|
type: 'error',
|
|
message: `Growth comment in r/${event.subreddit} was removed by Reddit`,
|
|
});
|
|
return;
|
|
case 'paused':
|
|
activityLogRepo.insert({
|
|
timestamp,
|
|
type: 'error',
|
|
message: `Account growth paused for u/${redditUsername}: ${event.reason}`,
|
|
});
|
|
return;
|
|
case 'trust-changed':
|
|
activityLogRepo.insert({
|
|
timestamp,
|
|
type: 'response',
|
|
message: `Growth allowance for u/${redditUsername} moved from level ${event.from} to `
|
|
+ `${event.to} (${event.rationale})`,
|
|
});
|
|
return;
|
|
}
|
|
}
|
|
|
|
function matrixRedditId(matrixUserId: string): string | null {
|
|
return matrixUserId.match(/^@t2_([^:]+):reddit\.com$/i)?.[1]?.toLowerCase() ?? null;
|
|
}
|
|
|
|
function delay(ms: number, signal: AbortSignal): Promise<void> {
|
|
if (signal.aborted) return Promise.reject(new DOMException('Operation aborted', 'AbortError'));
|
|
return new Promise((resolve, reject) => {
|
|
const onAbort = () => {
|
|
clearTimeout(timer);
|
|
signal.removeEventListener('abort', onAbort);
|
|
reject(new DOMException('Operation aborted', 'AbortError'));
|
|
};
|
|
const timer = setTimeout(() => {
|
|
signal.removeEventListener('abort', onAbort);
|
|
resolve();
|
|
}, ms);
|
|
signal.addEventListener('abort', onAbort, { once: true });
|
|
});
|
|
}
|