Pause suspended Reddit accounts
This commit is contained in:
1 parent
94049ed9c4
commit
7d039265e7
21 files changed
+359
-17
No files matched your search
@@ -68,6 +68,7 @@ SIGINT/SIGTERM abort shared loops, stop account runtimes, stop the web server, c
|
||||
- posts, messages, outcomes, activity, outbox, and conversation state repositories;
|
||||
- growth settings, community list, comment history, adaptive comment allowance, and isolated browsing policy/audit with optional submitted-vote state;
|
||||
- Matrix chat cache and sync cursor;
|
||||
- durable account suspension state; a Matrix `M_FORBIDDEN` or the periodic owned-profile check pauses only the affected runtime;
|
||||
- media index and files;
|
||||
- rate limiter and invitation cooldown;
|
||||
- response checker and conversation automation;
|
||||
|
||||
@@ -46,6 +46,8 @@ An account is eligible for new outreach only when its runtime is running, both i
|
||||
|
||||
Matrix cooldowns retain the operation that triggered them. A room-creation or invitation limit pauses new outreach but does not suppress replies in existing rooms; a limit returned while sending into an existing room pauses those sends as well.
|
||||
|
||||
An account-level Matrix `M_FORBIDDEN` is distinct from a room-level 403: it is treated as a definitive suspension signal. The account is durably marked suspended, its runtime is halted, and its historical content remains readable. Independently, every active account periodically reads its own Reddit profile; an explicit suspended profile result takes the same path. Missing profiles, network errors, and rate limits are not suspension evidence and do not pause an account.
|
||||
|
||||
The operator can manually clear the unsent funnel from the authenticated dashboard. The repository atomically moves discovered, screening, ready, and retryable candidates to a terminal rejection with the reason `Cleared manually by operator`, releases their leases, and retains the audit rows so those candidates are not rediscovered. A `sending` lease is never changed because its remote Matrix outcome may already be in flight. Clearing does not change bot mode; running discovery may refill the funnel immediately.
|
||||
|
||||
The same surface supports individual cancellation and prepared-message editing. Individual cancellation uses the same terminal audit transition and refuses `sending` or completed rows. Editing is limited to `ready` and `retryable` candidates with a preparation snapshot; it atomically replaces only the opener, clears generated strategy/experiment selection metadata, labels the model attribution `operator-edited`, and re-runs the deterministic outbound location/meeting safeguard. Both operations return HTTP 409 when a stale dashboard loses a race to dispatch.
|
||||
|
||||
+1
-1
@@ -10,7 +10,7 @@ The application uses one `better-sqlite3` database, normally `data/reddit-bot.db
|
||||
|
||||
### Account registry and account content
|
||||
|
||||
- `reddit_accounts`: verified Reddit/Matrix identity, encrypted-at-rest only to the degree provided by the host/volume, cookie JSON, Matrix credentials, active/status timestamps.
|
||||
- `reddit_accounts`: verified Reddit/Matrix identity, encrypted-at-rest only to the degree provided by the host/volume, cookie JSON, Matrix credentials, active/status timestamps. A confirmed suspension sets `status` to `suspended` and disables the account while retaining credentials and historical content for review.
|
||||
- `reddit_conversations`: account-scoped cached Matrix conversation payloads.
|
||||
- `reddit_account_sync_state`: account-scoped sync cursor, full/delta timestamps, and invitation cooldown.
|
||||
- `reddit_recipient_account_status`: process-global, case-insensitive recipient username status (`active`, `suspended`, or `deleted`) and last confirmed check time.
|
||||
|
||||
@@ -13,6 +13,7 @@ This is an ownership map, not an API dump. Read the owning implementation and te
|
||||
| `services/reddit-account-runtime-builder.ts` | Concrete account dependency graph and account loop | Account-specific browser, Matrix, repos, limiter, automation, chat, and media live here. |
|
||||
| `services/account-runtime-event-wiring.ts` | Per-account chat-to-realtime event bridge | Every account runtime publishes reply, media, and sync events into the shared dashboard/push stream; sync IDs remain account-scoped. |
|
||||
| `services/reddit-accounts.ts` | Account import/activation orchestration | Verification must bind Reddit cookies and Matrix identity before activation. |
|
||||
| `services/account-suspension.ts` | Owned-account suspension monitor | A definitive Matrix `M_FORBIDDEN` or Reddit suspended-profile result durably pauses only that account. |
|
||||
| `services/account-auth-context.ts` | Current verified identity snapshot | Account work pauses unless both sides authenticate. |
|
||||
| `services/session-bound-runtime.ts` | Legacy auth-bound background lifecycle | Prevents background work while logged out. |
|
||||
|
||||
|
||||
@@ -6,6 +6,7 @@ export const AttentionKindSchema = z.enum([
|
||||
'boundary',
|
||||
'failed-delivery',
|
||||
'account-login-required',
|
||||
'account-suspended',
|
||||
'runtime-failed',
|
||||
'invitation-cooldown',
|
||||
'account-unestablished',
|
||||
|
||||
@@ -31,7 +31,9 @@ export const RedditAccountSummarySchema = z.strictObject({
|
||||
redditUsername: z.string().min(1),
|
||||
matrixUserId: z.string().min(1),
|
||||
isRunning: z.boolean(),
|
||||
status: z.union([z.literal('authenticated'), z.literal('login-required'), z.literal('removed')]),
|
||||
status: z.union([
|
||||
z.literal('authenticated'), z.literal('login-required'), z.literal('suspended'), z.literal('removed'),
|
||||
]),
|
||||
createdAt: z.number(),
|
||||
updatedAt: z.number(),
|
||||
lastImportedAt: z.number(),
|
||||
|
||||
@@ -2,6 +2,8 @@ import type Database from 'better-sqlite3';
|
||||
|
||||
export type RedditAccountStatus =
|
||||
| 'verified'
|
||||
/** Reddit or Matrix confirmed that this account has been suspended. */
|
||||
| 'suspended'
|
||||
| 'login-required'
|
||||
| 'disabled'
|
||||
| 'legacy-unassigned'
|
||||
@@ -161,6 +163,19 @@ export class RedditAccountsRepository {
|
||||
`).run(now, accountId).changes === 1;
|
||||
}
|
||||
|
||||
/**
|
||||
* Disable an account only after Reddit or Matrix has returned a definitive
|
||||
* suspension signal. Credentials and historical content are retained for
|
||||
* operator review and a later verified session import.
|
||||
*/
|
||||
markSuspended(accountId: string, now: number = Date.now()): boolean {
|
||||
return this.db.prepare(`
|
||||
UPDATE reddit_accounts
|
||||
SET status = 'suspended', is_active = 0, updated_at = ?
|
||||
WHERE account_id = ? AND status NOT IN ('legacy-unassigned', 'removed')
|
||||
`).run(now, accountId).changes === 1;
|
||||
}
|
||||
|
||||
private getRequired(accountId: string): RedditAccountPrivateRecord {
|
||||
const account = this.get(accountId);
|
||||
if (!account) throw new Error(`Reddit account ${accountId} was not persisted`);
|
||||
|
||||
+35
-2
@@ -6,7 +6,7 @@ import fs from 'fs';
|
||||
export const LEGACY_UNASSIGNED_ACCOUNT_ID = '__legacy_unassigned__';
|
||||
|
||||
/** Migrations that replace a table other tables reference by foreign key. */
|
||||
const FOREIGN_KEY_REBUILD_VERSIONS = new Set([11, 32, 34]);
|
||||
const FOREIGN_KEY_REBUILD_VERSIONS = new Set([11, 32, 34, 35]);
|
||||
|
||||
export function createDatabase(config: Pick<Config, 'database'>): Database.Database {
|
||||
const dbPath = config.database.path === ':memory:'
|
||||
@@ -873,6 +873,39 @@ export function runMigrations(db: Database.Database): void {
|
||||
CREATE INDEX idx_learning_trajectories_account
|
||||
ON conversation_learning_trajectories(account_id, conversation_id, updated_at DESC);
|
||||
`),
|
||||
// A Matrix M_FORBIDDEN or owned-profile suspension must be a durable
|
||||
// account state, rather than a transient runtime failure.
|
||||
() => db.exec(`
|
||||
CREATE TABLE reddit_accounts_with_suspension (
|
||||
account_id TEXT PRIMARY KEY,
|
||||
reddit_username TEXT NOT NULL,
|
||||
matrix_user_id TEXT UNIQUE,
|
||||
status TEXT NOT NULL CHECK(status IN (
|
||||
'verified', 'login-required', 'disabled', 'suspended', 'legacy-unassigned', 'removed'
|
||||
)),
|
||||
is_active INTEGER NOT NULL DEFAULT 1 CHECK(is_active IN (0, 1)),
|
||||
cookies_json TEXT NOT NULL,
|
||||
matrix_credentials_json TEXT NOT NULL,
|
||||
created_at INTEGER NOT NULL,
|
||||
updated_at INTEGER NOT NULL,
|
||||
last_verified_at INTEGER,
|
||||
last_imported_at INTEGER,
|
||||
removed_at INTEGER
|
||||
);
|
||||
INSERT INTO reddit_accounts_with_suspension (
|
||||
account_id, reddit_username, matrix_user_id, status, is_active,
|
||||
cookies_json, matrix_credentials_json, created_at, updated_at,
|
||||
last_verified_at, last_imported_at, removed_at
|
||||
) SELECT
|
||||
account_id, reddit_username, matrix_user_id, status, is_active,
|
||||
cookies_json, matrix_credentials_json, created_at, updated_at,
|
||||
last_verified_at, last_imported_at, removed_at
|
||||
FROM reddit_accounts;
|
||||
DROP TABLE reddit_accounts;
|
||||
ALTER TABLE reddit_accounts_with_suspension RENAME TO reddit_accounts;
|
||||
CREATE INDEX idx_reddit_accounts_username ON reddit_accounts(reddit_username COLLATE NOCASE);
|
||||
CREATE INDEX idx_reddit_accounts_active ON reddit_accounts(is_active, status);
|
||||
`),
|
||||
];
|
||||
|
||||
let currentVersion = db.pragma('user_version', { simple: true }) as number;
|
||||
@@ -884,7 +917,7 @@ export function runMigrations(db: Database.Database): void {
|
||||
const version = index + 1;
|
||||
// SQLite cannot rebuild a referenced table while FK enforcement is on.
|
||||
// Migration 11 replaces the account-produced table graph atomically and
|
||||
// migration 32 rebuilds `reddit_accounts`, and migration 34 expands the
|
||||
// migrations 32 and 35 rebuild `reddit_accounts`, and migration 34 expands the
|
||||
// referenced learning-trajectory policy constraint; all validate the graph
|
||||
// before foreign-key enforcement is restored.
|
||||
const rebuildsForeignKeyGraph = FOREIGN_KEY_REBUILD_VERSIONS.has(version);
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
import type { Logger } from '../utils/logger.js';
|
||||
import { RedditRateLimitError, type RedditAccountProfile } from '../reddit/api.js';
|
||||
import { denyAccountAuth, type AccountAuthGate } from './account-auth-context.js';
|
||||
|
||||
export const ACCOUNT_SUSPENSION_CHECK_INTERVAL_MS = 6 * 60 * 60_000;
|
||||
const INITIAL_CHECK_DELAY_MS = 30_000;
|
||||
|
||||
export interface AccountSuspensionProfileLookup {
|
||||
getAccountProfile(username: string): Promise<RedditAccountProfile | null>;
|
||||
}
|
||||
|
||||
/**
|
||||
* Independently verifies that an owned Reddit account is still usable. This
|
||||
* is deliberately separate from establishment checks: disabling optional
|
||||
* establishment policy must never disable suspension protection.
|
||||
*/
|
||||
export class AccountSuspensionService {
|
||||
private intervalHandle: NodeJS.Timeout | null = null;
|
||||
private initialHandle: NodeJS.Timeout | null = null;
|
||||
private checking = false;
|
||||
|
||||
constructor(
|
||||
private readonly redditUsername: string,
|
||||
private readonly reddit: AccountSuspensionProfileLookup,
|
||||
private readonly pauseAccount: (reason: string) => void,
|
||||
private readonly logger: Logger,
|
||||
private readonly accountAuth: AccountAuthGate = denyAccountAuth,
|
||||
private readonly intervalMs = ACCOUNT_SUSPENSION_CHECK_INTERVAL_MS,
|
||||
) {}
|
||||
|
||||
start(): void {
|
||||
if (this.intervalHandle) return;
|
||||
this.initialHandle = setTimeout(() => {
|
||||
this.initialHandle = null;
|
||||
void this.checkNow();
|
||||
}, Math.min(INITIAL_CHECK_DELAY_MS, this.intervalMs));
|
||||
this.intervalHandle = setInterval(() => void this.checkNow(), this.intervalMs);
|
||||
this.logger.info(
|
||||
`Account suspension check for u/${this.redditUsername}:`
|
||||
+ ` scheduled every ${Math.round(this.intervalMs / 3_600_000)} hours`,
|
||||
);
|
||||
}
|
||||
|
||||
stop(): void {
|
||||
if (this.initialHandle) clearTimeout(this.initialHandle);
|
||||
if (this.intervalHandle) clearInterval(this.intervalHandle);
|
||||
this.initialHandle = null;
|
||||
this.intervalHandle = null;
|
||||
}
|
||||
|
||||
async checkNow(): Promise<boolean> {
|
||||
if (this.checking || !this.accountAuth.isFullyAuthenticated()) return false;
|
||||
this.checking = true;
|
||||
try {
|
||||
const profile = await this.reddit.getAccountProfile(this.redditUsername);
|
||||
if (profile?.isSuspended !== true) return false;
|
||||
const reason = 'Reddit profile marks this account as suspended';
|
||||
this.logger.error(`u/${this.redditUsername} is suspended; pausing the account`);
|
||||
this.pauseAccount(reason);
|
||||
return true;
|
||||
} catch (error) {
|
||||
if (error instanceof RedditRateLimitError) {
|
||||
this.logger.warn(`Account suspension check paused for u/${this.redditUsername}: Reddit rate-limited profile lookups`);
|
||||
return false;
|
||||
}
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
this.logger.warn(`Account suspension check failed for u/${this.redditUsername}: ${message}`);
|
||||
return false;
|
||||
} finally {
|
||||
this.checking = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -19,7 +19,7 @@ export type {
|
||||
ScrapedChatMessage,
|
||||
ScrapedRoom,
|
||||
} from './types.js';
|
||||
export { MatrixClient, MatrixAuthError } from './matrix/client.js';
|
||||
export { MatrixClient, MatrixAuthError, MatrixAccountSuspendedError } from './matrix/client.js';
|
||||
export type { MatrixCredentials } from './matrix/types.js';
|
||||
export { loadCachedCredentials, saveCredentials } from './matrix/credential-store.js';
|
||||
export {
|
||||
|
||||
@@ -14,6 +14,14 @@ export class MatrixAuthError extends Error {
|
||||
}
|
||||
}
|
||||
|
||||
/** Reddit's Matrix proxy uses this for an account that has been suspended. */
|
||||
export class MatrixAccountSuspendedError extends Error {
|
||||
constructor(message: string) {
|
||||
super(message);
|
||||
this.name = 'MatrixAccountSuspendedError';
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Thrown when Reddit's homeserver explicitly disables a Matrix endpoint.
|
||||
* Callers can remember the missing capability instead of retrying it for
|
||||
@@ -94,6 +102,8 @@ export interface MatrixClientOptions {
|
||||
requestTimeoutMs?: number;
|
||||
/** Test seam: inject a custom fetch (e.g. one that hangs). */
|
||||
fetchImpl?: typeof fetch;
|
||||
/** Durable account-level safety hook for a definitive M_FORBIDDEN result. */
|
||||
onAccountSuspended?: () => void;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -107,6 +117,7 @@ export class MatrixClient {
|
||||
private readonly minRequestIntervalMs: number;
|
||||
private readonly requestTimeoutMs: number;
|
||||
private readonly fetchImpl: typeof fetch;
|
||||
private readonly onAccountSuspended: (() => void) | undefined;
|
||||
|
||||
constructor(
|
||||
private creds: MatrixCredentials,
|
||||
@@ -116,6 +127,7 @@ export class MatrixClient {
|
||||
this.minRequestIntervalMs = options.minRequestIntervalMs ?? 30;
|
||||
this.requestTimeoutMs = options.requestTimeoutMs ?? 30_000;
|
||||
this.fetchImpl = options.fetchImpl ?? fetch;
|
||||
this.onAccountSuspended = options.onAccountSuspended;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -208,6 +220,12 @@ export class MatrixClient {
|
||||
|
||||
if (res.status === 401 || res.status === 403) {
|
||||
const failure = await readMatrixFailure(res);
|
||||
if (isAccountSuspendedFailure(failure.errcode)) {
|
||||
this.onAccountSuspended?.();
|
||||
throw new MatrixAccountSuspendedError(
|
||||
`Matrix ${method} ${path} → ${res.status} M_FORBIDDEN (account suspended)`,
|
||||
);
|
||||
}
|
||||
if (isMatrixAuthenticationFailure(res.status, failure.errcode)) {
|
||||
throw new MatrixAuthError(
|
||||
`Matrix ${method} ${path} → ${res.status} ${failure.errcode ?? ''} (token rejected)`,
|
||||
@@ -549,6 +567,10 @@ export class MatrixClient {
|
||||
if (response === null) throw new Error('Matrix media upload timed out');
|
||||
if (response.status === 401 || response.status === 403) {
|
||||
const failure = await readMatrixFailure(response);
|
||||
if (isAccountSuspendedFailure(failure.errcode)) {
|
||||
this.onAccountSuspended?.();
|
||||
throw new MatrixAccountSuspendedError('Matrix media upload → M_FORBIDDEN (account suspended)');
|
||||
}
|
||||
if (isMatrixAuthenticationFailure(response.status, failure.errcode)) {
|
||||
throw new MatrixAuthError(`Matrix media upload → ${response.status} (token rejected)`);
|
||||
}
|
||||
@@ -667,6 +689,10 @@ export class MatrixClient {
|
||||
}
|
||||
if (res.status === 401 || res.status === 403) {
|
||||
const failure = await readMatrixFailure(res);
|
||||
if (isAccountSuspendedFailure(failure.errcode)) {
|
||||
this.onAccountSuspended?.();
|
||||
throw new MatrixAccountSuspendedError('Matrix media download → M_FORBIDDEN (account suspended)');
|
||||
}
|
||||
if (isMatrixAuthenticationFailure(res.status, failure.errcode)) {
|
||||
throw new MatrixAuthError(
|
||||
`Matrix media download → ${res.status} ${failure.errcode ?? ''} (token rejected)`,
|
||||
@@ -724,6 +750,10 @@ function isMatrixAuthenticationFailure(status: number, errcode?: string): boolea
|
||||
return status === 401 || errcode === 'M_UNKNOWN_TOKEN' || errcode === 'M_MISSING_TOKEN';
|
||||
}
|
||||
|
||||
function isAccountSuspendedFailure(errcode?: string): boolean {
|
||||
return errcode === 'M_FORBIDDEN';
|
||||
}
|
||||
|
||||
function isFeatureDisabledFailure(responseText: string): boolean {
|
||||
try {
|
||||
const parsed = JSON.parse(responseText) as { message?: unknown };
|
||||
|
||||
@@ -70,6 +70,7 @@ import { AccountGrowthRepository } from '../database/growth-repository.js';
|
||||
import { AccountBrowsingSimulationRepository } from '../database/browsing-simulation-repository.js';
|
||||
import { CommentSourceWatchRepository } from '../database/comment-source-watch-repository.js';
|
||||
import { AccountBrowsingSimulationService } from './account-browsing-simulation.js';
|
||||
import { AccountSuspensionService } from './account-suspension.js';
|
||||
import { allowsConversationSync, allowsSending, type BotMode } from './bot-mode.js';
|
||||
import {
|
||||
wireAccountChatEvents,
|
||||
@@ -145,8 +146,17 @@ async function buildRuntime(
|
||||
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);
|
||||
let runtime: RedditAccountRuntime | null = null;
|
||||
const accountAuth = new AccountAuthContext(logger);
|
||||
const pauseSuspendedAccount = (reason: string): void => {
|
||||
accountsRepo.markSuspended(session.accountId);
|
||||
accountAuth?.invalidate(reason);
|
||||
runtime?.pause(reason);
|
||||
};
|
||||
const matrixClient = new MatrixClient(session.matrixCredentials, logger, {
|
||||
onAccountSuspended: () => pauseSuspendedAccount('Matrix returned M_FORBIDDEN: account suspended'),
|
||||
});
|
||||
const tokenRefresher = new BrowserMatrixTokenRefresher({
|
||||
context: browserClient.getContext()!,
|
||||
expectedUserId: session.matrixUserId,
|
||||
@@ -172,7 +182,6 @@ async function buildRuntime(
|
||||
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();
|
||||
@@ -274,6 +283,14 @@ async function buildRuntime(
|
||||
logger,
|
||||
accountAuth,
|
||||
);
|
||||
const suspensionService = new AccountSuspensionService(
|
||||
session.redditUsername,
|
||||
redditApi,
|
||||
pauseSuspendedAccount,
|
||||
logger,
|
||||
accountAuth,
|
||||
accountConfig.accountEstablishment.profileCheckIntervalMs,
|
||||
);
|
||||
const responseChecker = new ResponseCheckerService(
|
||||
chatPage, messagesRepo, successfulMessagesRepo, postsRepo, logger,
|
||||
accountAuth, conversationOutcomesRepo,
|
||||
@@ -392,12 +409,13 @@ async function buildRuntime(
|
||||
activityLogRepo,
|
||||
monitorService, processorService, responseChecker, conversationAutomation, chatService, mediaService,
|
||||
mediaSyncService, mediaExchange, rateLimiter,
|
||||
recipientAccountStatusService, establishmentService, growthService, browsingService,
|
||||
recipientAccountStatusService, establishmentService, suspensionService, growthService, browsingService,
|
||||
};
|
||||
const lifecycle = new AccountRuntimeLoop(
|
||||
resources, accountConfig, logger, getBotMode, recipientDeduplicator,
|
||||
);
|
||||
return new RedditAccountRuntime(session, resources, lifecycle);
|
||||
runtime = new RedditAccountRuntime(session, resources, lifecycle);
|
||||
return runtime;
|
||||
} catch (error) {
|
||||
await browserClient.close();
|
||||
throw error;
|
||||
@@ -436,6 +454,7 @@ export class AccountRuntimeLoop {
|
||||
this.controller = new AbortController();
|
||||
this.resources.recipientAccountStatusService.start();
|
||||
this.resources.establishmentService.start();
|
||||
this.resources.suspensionService?.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();
|
||||
@@ -448,12 +467,22 @@ export class AccountRuntimeLoop {
|
||||
this.resources.mediaSyncService.stop();
|
||||
this.resources.recipientAccountStatusService.stop();
|
||||
this.resources.establishmentService.stop();
|
||||
this.resources.suspensionService?.stop();
|
||||
await this.loop?.catch((error) => {
|
||||
if (!(error instanceof DOMException && error.name === 'AbortError')) throw error;
|
||||
});
|
||||
await this.resources.browserClient.close();
|
||||
}
|
||||
|
||||
pause(_reason: string): void {
|
||||
this.controller?.abort();
|
||||
this.mediaSyncRunning = false;
|
||||
this.resources.mediaSyncService.stop();
|
||||
this.resources.recipientAccountStatusService.stop();
|
||||
this.resources.establishmentService.stop();
|
||||
this.resources.suspensionService?.stop();
|
||||
}
|
||||
|
||||
/**
|
||||
* Starts or stops the background media sync only on a mode transition —
|
||||
* `MediaSyncService.start()` re-wraps its own `stop()` on every call, so
|
||||
|
||||
@@ -38,6 +38,7 @@ import type { RecipientAccountStatusService } from './recipient-account-status.j
|
||||
import type { AccountEstablishmentService } from './account-establishment.js';
|
||||
import type { AccountGrowthService } from './account-growth.js';
|
||||
import type { AccountBrowsingSimulationService } from './account-browsing-simulation.js';
|
||||
import type { AccountSuspensionService } from './account-suspension.js';
|
||||
import type { ProcessedPost, ProcessResult } from './processor.js';
|
||||
import type { ConversationEventRepository } from '../database/conversation-event-repository.js';
|
||||
import type { ConversationLearningRepository } from '../database/conversation-learning-repository.js';
|
||||
@@ -91,6 +92,8 @@ export interface RedditAccountRuntimeResources {
|
||||
mediaSyncService: MediaSyncService;
|
||||
recipientAccountStatusService: RecipientAccountStatusService;
|
||||
establishmentService: AccountEstablishmentService;
|
||||
/** Present in concrete runtimes; optional for narrow lifecycle test fixtures. */
|
||||
suspensionService?: AccountSuspensionService;
|
||||
/**
|
||||
* Comment-based account growth. Optional because a runtime is fully
|
||||
* functional without it, and it is disabled per account by default.
|
||||
@@ -107,6 +110,8 @@ export interface RedditAccountRuntimeLifecycle {
|
||||
start(): void | Promise<void>;
|
||||
/** Stop and await this account's schedules/loop, then close its browser. */
|
||||
stop(): void | Promise<void>;
|
||||
/** Immediately halt account work after a definitive suspension signal. */
|
||||
pause?(reason: string): void;
|
||||
/** Process a post assigned by the shared discovery scheduler. */
|
||||
dispatchPost?(post: RedditPost, signal?: AbortSignal): Promise<void>;
|
||||
preparePost?(post: RedditPost, signal?: AbortSignal): Promise<ProcessResult>;
|
||||
@@ -123,6 +128,7 @@ export type RedditAccountRuntimeState =
|
||||
| 'starting'
|
||||
| 'running'
|
||||
| 'stopping'
|
||||
| 'paused'
|
||||
| 'stopped'
|
||||
| 'failed';
|
||||
|
||||
@@ -197,6 +203,17 @@ export class RedditAccountRuntime {
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Suspension can be discovered by work that this runtime is currently
|
||||
* awaiting. Keep this synchronous so it cannot wait on its own loop.
|
||||
*/
|
||||
pause(reason: string): void {
|
||||
if (this.state === 'stopped' || this.state === 'failed') return;
|
||||
this.failure = reason;
|
||||
this.state = 'paused';
|
||||
this.lifecycle.pause?.(reason);
|
||||
}
|
||||
|
||||
snapshot(): RedditAccountRuntimeSnapshot {
|
||||
return {
|
||||
accountId: this.session.accountId,
|
||||
@@ -294,7 +311,9 @@ export class RedditAccountRuntimeManager {
|
||||
|
||||
async startEnabled(): Promise<RedditAccountRuntimeManagerSnapshot> {
|
||||
const enabled = this.accounts.listPublic().filter(isEnabledAccount);
|
||||
await Promise.all(enabled.map((account) => this.start(account.accountId)));
|
||||
// A broken or suspended session belongs to one account. Do not prevent
|
||||
// other verified accounts from starting because of it.
|
||||
await Promise.allSettled(enabled.map((account) => this.start(account.accountId)));
|
||||
return this.snapshot();
|
||||
}
|
||||
|
||||
|
||||
@@ -51,7 +51,7 @@ export interface RedditAccountSummary {
|
||||
matrixUserId: string;
|
||||
isRunning: boolean;
|
||||
/** `removed` is terminal until a fresh session is imported for the account. */
|
||||
status: 'authenticated' | 'login-required' | 'removed';
|
||||
status: 'authenticated' | 'login-required' | 'suspended' | 'removed';
|
||||
createdAt: number;
|
||||
updatedAt: number;
|
||||
lastImportedAt: number;
|
||||
@@ -136,16 +136,20 @@ export class RedditAccountsService {
|
||||
|
||||
private toSummary(account: RedditAccountPublicRecord): RedditAccountSummary {
|
||||
const runtime = this.runtimes?.get(account.accountId);
|
||||
const running = account.status === 'removed' ? false : runtime?.snapshot().state === 'running';
|
||||
const running = account.status === 'removed' || account.status === 'suspended'
|
||||
? false
|
||||
: runtime?.snapshot().state === 'running';
|
||||
return {
|
||||
accountId: account.accountId,
|
||||
redditUsername: account.redditUsername,
|
||||
matrixUserId: account.matrixUserId ?? account.accountId,
|
||||
isRunning: account.status === 'removed'
|
||||
isRunning: account.status === 'removed' || account.status === 'suspended'
|
||||
? false
|
||||
: running ?? (account.accountId === this.auth.getSnapshot().accountId),
|
||||
status: account.status === 'removed'
|
||||
? 'removed'
|
||||
: account.status === 'suspended'
|
||||
? 'suspended'
|
||||
: account.status === 'verified' && (running ?? account.accountId === this.auth.getSnapshot().accountId)
|
||||
? 'authenticated'
|
||||
: 'login-required',
|
||||
|
||||
@@ -3,7 +3,7 @@ import type { WebContext } from '../context.js';
|
||||
|
||||
interface AttentionItem {
|
||||
id: string;
|
||||
kind: 'human-required' | 'boundary' | 'failed-delivery' | 'account-login-required' | 'runtime-failed' | 'invitation-cooldown' | 'account-unestablished';
|
||||
kind: 'human-required' | 'boundary' | 'failed-delivery' | 'account-login-required' | 'account-suspended' | 'runtime-failed' | 'invitation-cooldown' | 'account-unestablished';
|
||||
severity: 'warning' | 'critical';
|
||||
title: string;
|
||||
detail: string;
|
||||
@@ -38,7 +38,7 @@ export function createAttentionRouter(ctx: WebContext): Router {
|
||||
owner,
|
||||
});
|
||||
}
|
||||
if (account.status !== 'login-required') continue;
|
||||
if (account.status === 'login-required') {
|
||||
items.push({
|
||||
id: `account:${account.accountId}`,
|
||||
kind: 'account-login-required',
|
||||
@@ -49,6 +49,18 @@ export function createAttentionRouter(ctx: WebContext): Router {
|
||||
owner,
|
||||
});
|
||||
}
|
||||
if (account.status === 'suspended') {
|
||||
items.push({
|
||||
id: `account:${account.accountId}`,
|
||||
kind: 'account-suspended',
|
||||
severity: 'critical',
|
||||
title: `u/${account.redditUsername} is suspended`,
|
||||
detail: 'Reddit or Matrix confirmed suspension, so this account has been paused.',
|
||||
occurredAt: account.updatedAt,
|
||||
owner,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
const snapshots = ctx.services.accountRuntimes?.list() ?? [];
|
||||
for (const snapshot of snapshots) {
|
||||
|
||||
@@ -56,6 +56,8 @@ export function createStatsRouter(ctx: WebContext): Router {
|
||||
const runtimeState = snapshot?.state ?? 'stopped';
|
||||
const sendingState: BotStats['accounts'][number]['sendingState'] = runtimeState === 'failed'
|
||||
? 'runtime-failed'
|
||||
: account.status === 'suspended'
|
||||
? 'account-suspended'
|
||||
: account.status === 'login-required' || (runtimeState === 'running' && !authenticated)
|
||||
? 'login-required'
|
||||
: runtimeState !== 'running'
|
||||
|
||||
+2
-2
@@ -15,9 +15,9 @@ export interface BotStats {
|
||||
accounts: Array<{
|
||||
accountId: string;
|
||||
redditUsername: string;
|
||||
runtimeState: 'created' | 'starting' | 'running' | 'stopping' | 'stopped' | 'failed';
|
||||
runtimeState: 'created' | 'starting' | 'running' | 'stopping' | 'paused' | 'stopped' | 'failed';
|
||||
authenticated: boolean;
|
||||
sendingState: 'ready' | 'bot-stopped' | 'scan-only' | 'runtime-stopped' | 'runtime-failed' | 'login-required' | 'matrix-cooldown' | 'replies-prioritized' | 'pacing' | 'unestablished';
|
||||
sendingState: 'ready' | 'bot-stopped' | 'scan-only' | 'runtime-stopped' | 'runtime-failed' | 'login-required' | 'account-suspended' | 'matrix-cooldown' | 'replies-prioritized' | 'pacing' | 'unestablished';
|
||||
messagesThisHour: number;
|
||||
maxMessagesPerHour: number;
|
||||
nextMessageAvailableAt: number | null;
|
||||
|
||||
@@ -95,6 +95,17 @@ describe('multi-account database ownership', () => {
|
||||
expect(accounts.get('t2_bob')).toMatchObject({ status: 'disabled', isActive: false, updatedAt: 130 });
|
||||
});
|
||||
|
||||
it('durably pauses a suspended account without destroying its session or history', () => {
|
||||
addAccount('t2_alice', 'alice-bot', 100);
|
||||
|
||||
expect(accounts.markSuspended('t2_alice', 200)).toBe(true);
|
||||
expect(accounts.get('t2_alice')).toMatchObject({
|
||||
status: 'suspended', isActive: false, updatedAt: 200,
|
||||
cookiesJson: expect.stringContaining('alice-bot-secret'),
|
||||
});
|
||||
expect(accounts.getActive()).toEqual([]);
|
||||
});
|
||||
|
||||
it('partitions conversations, sync state and media even when remote IDs match', () => {
|
||||
addAccount('t2_alice', 'alice-bot', 100);
|
||||
addAccount('t2_bob', 'bob-bot', 110);
|
||||
|
||||
@@ -0,0 +1,49 @@
|
||||
import { describe, expect, it, vi } from 'vitest';
|
||||
import { AccountSuspensionService } from '../src/services/account-suspension.js';
|
||||
import { RedditRateLimitError } from '../src/reddit/api.js';
|
||||
import { createMockLogger } from './mockLogger.js';
|
||||
|
||||
const activeProfile = {
|
||||
createdAt: null, linkKarma: null, commentKarma: null, totalKarma: null,
|
||||
hasVerifiedEmail: null, isSuspended: false,
|
||||
};
|
||||
|
||||
describe('AccountSuspensionService', () => {
|
||||
it('pauses an authenticated account only when its own profile confirms suspension', async () => {
|
||||
const pause = vi.fn();
|
||||
const reddit = { getAccountProfile: vi.fn().mockResolvedValue({ ...activeProfile, isSuspended: true }) };
|
||||
const service = new AccountSuspensionService(
|
||||
'owned-account', reddit, pause, createMockLogger(), { isFullyAuthenticated: () => true },
|
||||
);
|
||||
|
||||
await expect(service.checkNow()).resolves.toBe(true);
|
||||
expect(reddit.getAccountProfile).toHaveBeenCalledWith('owned-account');
|
||||
expect(pause).toHaveBeenCalledWith('Reddit profile marks this account as suspended');
|
||||
});
|
||||
|
||||
it('does not pause for a missing profile, a transient failure, or a rate limit', async () => {
|
||||
const pause = vi.fn();
|
||||
const reddit = { getAccountProfile: vi.fn()
|
||||
.mockResolvedValueOnce(null)
|
||||
.mockRejectedValueOnce(new Error('upstream unavailable'))
|
||||
.mockRejectedValueOnce(new RedditRateLimitError(1_000)) };
|
||||
const service = new AccountSuspensionService(
|
||||
'owned-account', reddit, pause, createMockLogger(), { isFullyAuthenticated: () => true },
|
||||
);
|
||||
|
||||
await expect(service.checkNow()).resolves.toBe(false);
|
||||
await expect(service.checkNow()).resolves.toBe(false);
|
||||
await expect(service.checkNow()).resolves.toBe(false);
|
||||
expect(pause).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('does not probe while the account is not authenticated', async () => {
|
||||
const reddit = { getAccountProfile: vi.fn() };
|
||||
const service = new AccountSuspensionService(
|
||||
'owned-account', reddit, vi.fn(), createMockLogger(), { isFullyAuthenticated: () => false },
|
||||
);
|
||||
|
||||
await expect(service.checkNow()).resolves.toBe(false);
|
||||
expect(reddit.getAccountProfile).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@@ -1,6 +1,7 @@
|
||||
import { describe, it, expect, vi } from 'vitest';
|
||||
import {
|
||||
MatrixAuthError,
|
||||
MatrixAccountSuspendedError,
|
||||
MatrixClient,
|
||||
MatrixFeatureDisabledError,
|
||||
MatrixRateLimitError,
|
||||
@@ -186,6 +187,20 @@ describe('MatrixClient direct room registration', () => {
|
||||
});
|
||||
|
||||
describe('MatrixClient authentication classification', () => {
|
||||
it('calls the account pause hook and throws a typed error for M_FORBIDDEN', async () => {
|
||||
const onAccountSuspended = vi.fn();
|
||||
const client = new MatrixClient(creds, createMockLogger(), {
|
||||
minRequestIntervalMs: 0,
|
||||
onAccountSuspended,
|
||||
fetchImpl: vi.fn().mockResolvedValue(new Response(JSON.stringify({
|
||||
errcode: 'M_FORBIDDEN', error: 'User is suspended',
|
||||
}), { status: 403, headers: { 'content-type': 'application/json' } })),
|
||||
});
|
||||
|
||||
await expect(client.whoami()).rejects.toBeInstanceOf(MatrixAccountSuspendedError);
|
||||
expect(onAccountSuspended).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it('classifies an explicitly disabled endpoint as an unavailable feature', async () => {
|
||||
const client = new MatrixClient(creds, createMockLogger(), {
|
||||
minRequestIntervalMs: 0,
|
||||
|
||||
@@ -103,6 +103,28 @@ describe('RedditAccountRuntimeManager', () => {
|
||||
expect(manager.list()).toHaveLength(1);
|
||||
});
|
||||
|
||||
it('keeps healthy accounts running when one enabled account cannot start', async () => {
|
||||
const broken = account('broken');
|
||||
const healthy = account('healthy');
|
||||
const store = accountStore([broken, healthy]);
|
||||
const factory = new RedditAccountRuntimeFactory(async (session) => new RedditAccountRuntime(
|
||||
session,
|
||||
{ ...resources },
|
||||
{
|
||||
start: () => {
|
||||
if (session.accountId === broken.accountId) throw new Error('Matrix account suspended');
|
||||
},
|
||||
stop: vi.fn(),
|
||||
},
|
||||
));
|
||||
const manager = new RedditAccountRuntimeManager(store.repository, factory);
|
||||
|
||||
await manager.startEnabled();
|
||||
|
||||
expect(manager.get(broken.accountId)).toBeNull();
|
||||
expect(manager.get(healthy.accountId)?.snapshot().state).toBe('running');
|
||||
});
|
||||
|
||||
it('replaces only the refreshed account while other accounts keep running', async () => {
|
||||
const first = account('first');
|
||||
const second = account('second');
|
||||
@@ -220,6 +242,27 @@ describe('RedditAccountRuntimeManager', () => {
|
||||
});
|
||||
|
||||
describe('account runtime isolation', () => {
|
||||
it('pauses an account immediately and refuses new work', async () => {
|
||||
const stored = account('suspended');
|
||||
const pause = vi.fn();
|
||||
const runtime = new RedditAccountRuntime(
|
||||
{
|
||||
accountId: stored.accountId, redditUsername: stored.redditUsername,
|
||||
matrixUserId: stored.matrixUserId!, cookies: [],
|
||||
matrixCredentials: JSON.parse(stored.matrixCredentialsJson), updatedAt: stored.updatedAt,
|
||||
},
|
||||
{ ...resources },
|
||||
{ start: vi.fn(), stop: vi.fn(), pause },
|
||||
);
|
||||
await runtime.start();
|
||||
|
||||
runtime.pause('Matrix returned M_FORBIDDEN: account suspended');
|
||||
|
||||
expect(runtime.snapshot()).toMatchObject({ state: 'paused', failure: expect.stringContaining('M_FORBIDDEN') });
|
||||
expect(pause).toHaveBeenCalledOnce();
|
||||
expect(await runtime.canScan()).toBe(false);
|
||||
});
|
||||
|
||||
it('does not accept new outreach while the account has actionable replies', async () => {
|
||||
const stored = account('reply_first');
|
||||
const hasActionableReplies = vi.fn().mockResolvedValue(true);
|
||||
|
||||
Reference in new issue
Block a user