feat: replace browser login with imported Reddit sessions
This commit is contained in:
1 parent
2b0545afcf
commit
c483c7b8be
76 files changed
+2048
-2828
No files matched your search
+100
-26
@@ -2,9 +2,66 @@ import type { AppContainer } from './bootstrap.js';
|
||||
import { withRetry } from './utils/retry.js';
|
||||
import { isSkip } from './services/processor.js';
|
||||
import { reportDeliveryResult, reportInvitationCooldown } from './web/event-reporting.js';
|
||||
import { SessionBoundRuntime } from './services/session-bound-runtime.js';
|
||||
|
||||
function sleep(ms: number): Promise<void> {
|
||||
return new Promise((resolve) => setTimeout(resolve, ms));
|
||||
export interface StoppableSchedule {
|
||||
stop(): void;
|
||||
}
|
||||
|
||||
/** The limiter increments after a successful send, so its count is already current. */
|
||||
export function currentHourlySendCount(
|
||||
limiter: { getStats(): { messagesThisHour: number } },
|
||||
): number {
|
||||
return limiter.getStats().messagesThisHour;
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs a task again only after its previous run has settled. This deliberately
|
||||
* replaces setInterval: a slow browser response check must never overlap the
|
||||
* next check and contend for the same browser/session resources.
|
||||
*/
|
||||
export function startNonOverlappingSchedule(
|
||||
task: () => Promise<void>,
|
||||
intervalMs: number,
|
||||
): StoppableSchedule {
|
||||
let stopped = false;
|
||||
let timer: NodeJS.Timeout | null = null;
|
||||
|
||||
const run = async (): Promise<void> => {
|
||||
if (stopped) return;
|
||||
try {
|
||||
await task();
|
||||
} finally {
|
||||
if (!stopped) timer = setTimeout(() => void run(), intervalMs);
|
||||
}
|
||||
};
|
||||
|
||||
void run();
|
||||
return {
|
||||
stop: () => {
|
||||
stopped = true;
|
||||
if (timer) clearTimeout(timer);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function sleep(ms: number, signal?: AbortSignal): Promise<void> {
|
||||
if (signal?.aborted) return Promise.reject(new DOMException('Operation aborted', 'AbortError'));
|
||||
return new Promise((resolve, reject) => {
|
||||
let timer: NodeJS.Timeout;
|
||||
const abort = (): void => {
|
||||
clearTimeout(timer);
|
||||
cleanup();
|
||||
reject(new DOMException('Operation aborted', 'AbortError'));
|
||||
};
|
||||
const cleanup = (): void => signal?.removeEventListener('abort', abort);
|
||||
const done = (): void => {
|
||||
cleanup();
|
||||
resolve();
|
||||
};
|
||||
timer = setTimeout(done, ms);
|
||||
signal?.addEventListener('abort', abort, { once: true });
|
||||
});
|
||||
}
|
||||
|
||||
const PAUSED_LOG_INTERVAL_MS = 60_000;
|
||||
@@ -20,7 +77,6 @@ export async function runMainLoop(app: AppContainer): Promise<void> {
|
||||
config,
|
||||
logger,
|
||||
browserClient,
|
||||
accountAuth,
|
||||
postsRepo,
|
||||
messagesRepo,
|
||||
redditApi,
|
||||
@@ -35,21 +91,22 @@ export async function runMainLoop(app: AppContainer): Promise<void> {
|
||||
} = app;
|
||||
|
||||
let isShuttingDown = false;
|
||||
let responseCheckInterval: NodeJS.Timeout | null = null;
|
||||
const shutdownController = new AbortController();
|
||||
let responseCheckSchedule: StoppableSchedule | null = null;
|
||||
let sessionRuntime: SessionBoundRuntime | null = null;
|
||||
|
||||
const shutdown = async (): Promise<void> => {
|
||||
if (isShuttingDown) return;
|
||||
isShuttingDown = true;
|
||||
shutdownController.abort();
|
||||
|
||||
if (responseCheckInterval) clearInterval(responseCheckInterval);
|
||||
mediaSyncService.stop();
|
||||
sessionRuntime?.shutdown();
|
||||
|
||||
logger.info('Shutting down gracefully...');
|
||||
webServer?.stop();
|
||||
await browserClient.close();
|
||||
db.close();
|
||||
logger.info('Goodbye!');
|
||||
process.exit(0);
|
||||
};
|
||||
|
||||
process.on('SIGINT', shutdown);
|
||||
@@ -67,7 +124,7 @@ export async function runMainLoop(app: AppContainer): Promise<void> {
|
||||
const RESPONSE_CHECK_INTERVAL_MS = Math.max(config.reddit.pollIntervalMs * 5, 60_000);
|
||||
|
||||
const checkResponses = async (): Promise<void> => {
|
||||
if (isShuttingDown || !accountAuth.isFullyAuthenticated()) return;
|
||||
if (isShuttingDown) return;
|
||||
try {
|
||||
logger.debug('Checking for responses to sent messages (Background Thread)...');
|
||||
const responses = await responseChecker.checkForResponses();
|
||||
@@ -90,36 +147,51 @@ export async function runMainLoop(app: AppContainer): Promise<void> {
|
||||
}
|
||||
};
|
||||
|
||||
checkResponses();
|
||||
responseCheckInterval = setInterval(checkResponses, RESPONSE_CHECK_INTERVAL_MS);
|
||||
let startupSyncStarted = false;
|
||||
sessionRuntime = new SessionBoundRuntime(
|
||||
() => {
|
||||
responseCheckSchedule = startNonOverlappingSchedule(checkResponses, RESPONSE_CHECK_INTERVAL_MS);
|
||||
if (config.media.syncEnabled) mediaSyncService.start();
|
||||
if (!startupSyncStarted) {
|
||||
startupSyncStarted = true;
|
||||
chatService
|
||||
.runStartupSync()
|
||||
.then(async ({ kind, chats }) => {
|
||||
logger.info(`[chat] startup sync (${kind}) complete: ${chats} chats`);
|
||||
await chatService.enrichUsernames();
|
||||
})
|
||||
.catch((err) => logger.error('[chat] startup sync failed:', err));
|
||||
}
|
||||
},
|
||||
() => {
|
||||
responseCheckSchedule?.stop();
|
||||
responseCheckSchedule = null;
|
||||
mediaSyncService.stop();
|
||||
},
|
||||
);
|
||||
|
||||
let lastPausedLog = 0;
|
||||
let lastAuthPausedLog = 0;
|
||||
let lastCooldownLog = 0;
|
||||
let reportedCooldownUntil: number | null = null;
|
||||
const COOLDOWN_LOG_INTERVAL_MS = 5 * 60_000;
|
||||
|
||||
while (!isShuttingDown) {
|
||||
try {
|
||||
if (!accountAuth.isFullyAuthenticated()) {
|
||||
await accountAuth.ensureFullyAuthenticated();
|
||||
}
|
||||
if (!accountAuth.isFullyAuthenticated()) {
|
||||
if (Date.now() - lastAuthPausedLog > PAUSED_LOG_INTERVAL_MS) {
|
||||
logger.info('Bot logic is locked until full Reddit and Matrix authentication is verified.');
|
||||
lastAuthPausedLog = Date.now();
|
||||
if (!sessionRuntime.reconcile(browserClient.getSessionState())) {
|
||||
if (Date.now() - lastPausedLog > PAUSED_LOG_INTERVAL_MS) {
|
||||
logger.info('Reddit is logged out. Waiting for a session dump from the extension.');
|
||||
lastPausedLog = Date.now();
|
||||
}
|
||||
await sleep(PAUSE_POLL_MS);
|
||||
await sleep(PAUSE_POLL_MS, shutdownController.signal);
|
||||
continue;
|
||||
}
|
||||
lastAuthPausedLog = 0;
|
||||
|
||||
if (webServer && !webServer.isRunning()) {
|
||||
if (Date.now() - lastPausedLog > PAUSED_LOG_INTERVAL_MS) {
|
||||
logger.info('Bot is paused. Use the web dashboard to start.');
|
||||
lastPausedLog = Date.now();
|
||||
}
|
||||
await sleep(PAUSE_POLL_MS);
|
||||
await sleep(PAUSE_POLL_MS, shutdownController.signal);
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -140,7 +212,7 @@ export async function runMainLoop(app: AppContainer): Promise<void> {
|
||||
);
|
||||
lastCooldownLog = Date.now();
|
||||
}
|
||||
await sleep(config.reddit.pollIntervalMs);
|
||||
await sleep(config.reddit.pollIntervalMs, shutdownController.signal);
|
||||
continue;
|
||||
}
|
||||
if (webServer && reportedCooldownUntil !== null) {
|
||||
@@ -254,8 +326,7 @@ export async function runMainLoop(app: AppContainer): Promise<void> {
|
||||
strategyId: processed.strategyId,
|
||||
});
|
||||
|
||||
const stats = rateLimiter.getStats();
|
||||
const currentCount = stats.messagesThisHour + 1;
|
||||
const currentCount = currentHourlySendCount(rateLimiter);
|
||||
logger.info(
|
||||
`Message sent successfully! (${currentCount}/${config.rateLimit.maxMessagesPerHour} this hour)`,
|
||||
);
|
||||
@@ -297,8 +368,10 @@ export async function runMainLoop(app: AppContainer): Promise<void> {
|
||||
return success;
|
||||
},
|
||||
(success) => success === true,
|
||||
shutdownController.signal,
|
||||
);
|
||||
} catch (error) {
|
||||
if (isShuttingDown && error instanceof DOMException && error.name === 'AbortError') break;
|
||||
const errorMessage = error instanceof Error ? error.message : String(error);
|
||||
logger.error(`Error processing post ${post.id}:`, error);
|
||||
postsRepo.setError(post.id, errorMessage);
|
||||
@@ -307,11 +380,12 @@ export async function runMainLoop(app: AppContainer): Promise<void> {
|
||||
|
||||
if (!isShuttingDown) {
|
||||
logger.debug(`Waiting ${config.reddit.pollIntervalMs / 1000} seconds before next poll...`);
|
||||
await sleep(config.reddit.pollIntervalMs);
|
||||
await sleep(config.reddit.pollIntervalMs, shutdownController.signal);
|
||||
}
|
||||
} catch (error) {
|
||||
if (isShuttingDown && error instanceof DOMException && error.name === 'AbortError') break;
|
||||
logger.error('Error in main loop:', error);
|
||||
await sleep(ERROR_BACKOFF_MS);
|
||||
await sleep(ERROR_BACKOFF_MS, shutdownController.signal);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user