jjbrunton/RedditBot
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: support concurrent Reddit accounts

Browse Source
This commit is contained in:
jjbrunton committed 2026-08-08 19:00:38 +01:00
1 parent d819a0f1e6
commit 16a31335d0
51 files changed
+3241 -250
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/bootstrap.ts
+89 -11
View File
@@ -39,6 +39,13 @@ import { reportReply } from './web/event-reporting.js';
import { RedditSessionImporter } from './services/reddit-session-importer.js';
import { AccountAuthContext } from './services/account-auth-context.js';
import { ConfigStore } from './config/persistence.js';
import { RedditAccountsRepository, AccountContentRepository } from './database/account-repositories.js';
import { RedditAccountsService } from './services/reddit-accounts.js';
import type { Cookie } from 'playwright';
import { ChatCache } from './services/chat/cache.js';
import { RedditAccountRuntimeManager } from './services/reddit-account-runtime.js';
import { createConcreteRedditAccountRuntimeFactory } from './services/reddit-account-runtime-builder.js';
import { importLegacyAccountContent } from './database/legacy-content-import.js';
export interface AppContainer {
config: Config;
@@ -47,6 +54,8 @@ export interface AppContainer {
browserClient: BrowserClient;
accountAuth: AccountAuthContext;
redditAccountsRepo: RedditAccountsRepository;
accountRuntimeManager: RedditAccountRuntimeManager;
postsRepo: PostsRepository;
messagesRepo: MessagesRepository;
@@ -97,14 +106,25 @@ export async function bootstrap(
// Database
const db = existingDatabase ?? createDatabase(config);
runMigrations(db);
const postsRepo = new PostsRepository(db);
const messagesRepo = new MessagesRepository(db);
const successfulMessagesRepo = new SuccessfulMessagesRepository(db);
const activityLogRepo = new ActivityLogRepository(db);
const bootstrapAccountsRepo = new RedditAccountsRepository(db);
const bootstrapAccount = bootstrapAccountsRepo.listPublic()
.filter((account) => account.status === 'verified')
.sort((left, right) => right.updatedAt - left.updatedAt)[0];
const bootstrapAccountId = bootstrapAccount?.accountId;
const postsRepo = new PostsRepository(db, bootstrapAccountId);
const messagesRepo = new MessagesRepository(db, bootstrapAccountId);
const successfulMessagesRepo = new SuccessfulMessagesRepository(db, bootstrapAccountId);
const activityLogRepo = new ActivityLogRepository(db, bootstrapAccountId);
const redditUsersRepo = new RedditUsersRepository(db);
const deliveryOutboxRepo = new DeliveryOutboxRepository(db);
const conversationStateRepo = new ConversationStateRepository(db);
const conversationOutcomesRepo = new ConversationOutcomesRepository(db);
const deliveryOutboxRepo = new DeliveryOutboxRepository(db, bootstrapAccountId);
const conversationStateRepo = new ConversationStateRepository(db, bootstrapAccountId);
const conversationOutcomesRepo = new ConversationOutcomesRepository(db, bootstrapAccountId);
const redditAccountsRepo = bootstrapAccountsRepo;
const accountContentRepo = new AccountContentRepository(db);
const legacyImport = importLegacyAccountContent(db);
if (legacyImport.conversations > 0 || legacyImport.media > 0) {
logger.info(`Imported legacy unassigned content into SQLite (${legacyImport.conversations} conversations, ${legacyImport.media} media items)`);
}
logger.info('Database initialized');
// Browser
@@ -112,6 +132,18 @@ export async function bootstrap(
const accountAuth = new AccountAuthContext(logger);
await browserClient.initialize();
const storedAccount = bootstrapAccount;
if (storedAccount) {
const privateAccount = redditAccountsRepo.get(storedAccount.accountId);
if (privateAccount) {
try {
await browserClient.importRedditCookies(JSON.parse(privateAccount.cookiesJson) as Cookie[]);
} catch (error) {
logger.warn(`Could not restore stored Reddit cookies for u/${storedAccount.redditUsername}:`, error);
}
}
}
// The extension import is the only session acquisition path. A missing
// session is recoverable: the dashboard remains available while all
// account-bound work waits for a verified dump.
@@ -151,7 +183,10 @@ export async function bootstrap(
// Matrix credentials come only from a verified extension dump (or its
// persisted cache on restart). The backend never opens Reddit Chat to
// sniff or refresh a token.
let matrixCreds: MatrixCredentials | null = await loadCachedCredentials();
const storedPrivateAccount = storedAccount ? redditAccountsRepo.get(storedAccount.accountId) : null;
let matrixCreds: MatrixCredentials | null = storedPrivateAccount
? JSON.parse(storedPrivateAccount.matrixCredentialsJson) as MatrixCredentials
: await loadCachedCredentials();
if (matrixCreds) {
logger.info(`Loaded cached Matrix credentials (user=${matrixCreds.userId})`);
}
@@ -185,7 +220,13 @@ export async function bootstrap(
// 2. Build the media downloader + service on top of the Matrix client so
// `mxc://` URIs resolve through the same auth + rate-limiter pipeline.
const mediaDownloader = new MatrixMediaDownloader(activeMatrixClient, logger);
const mediaService = new MediaService(mediaDownloader, logger);
const mediaService = new MediaService(
mediaDownloader,
logger,
undefined,
accountContentRepo,
() => accountAuth.getSnapshot().accountId,
);
const indexedMedia = await mediaService.getMediaItems();
const receivedImageRooms = new Set(
@@ -224,7 +265,7 @@ export async function bootstrap(
mediaService,
tokenRefresher,
usernameResolver,
undefined,
new ChatCache(logger, undefined, accountContentRepo, storedAccount?.accountId ?? null),
messagesRepo,
redditUsersRepo,
accountAuth,
@@ -247,13 +288,42 @@ export async function bootstrap(
mediaDownloader.setMatrixClient(client);
matrixUsernameResolver.setMatrixClient(client);
},
switchAccount(accountId) {
return chatService.switchAccount(accountId);
},
},
(credentials) => new MatrixClient(credentials, logger),
saveCredentials,
accountAuth,
{
save(input) {
redditAccountsRepo.upsertVerified({
accountId: input.accountId,
redditUsername: input.redditUsername,
matrixUserId: input.matrixUserId,
cookiesJson: JSON.stringify(input.cookies),
matrixCredentialsJson: JSON.stringify(input.credentials),
});
},
},
);
let webServer: WebServer | null = null;
const accountRuntimeManager = new RedditAccountRuntimeManager(
redditAccountsRepo,
createConcreteRedditAccountRuntimeFactory({
config,
logger,
db,
isBotRunning: () => webServer?.isRunning() ?? true,
}),
);
const redditAccounts = new RedditAccountsService(
redditAccountsRepo,
sessionImporter,
accountAuth,
accountRuntimeManager,
browserClient,
);
if (config.web.enabled) {
webServer = new WebServer(
config.web.port,
@@ -330,6 +400,8 @@ export async function bootstrap(
webServer.setBrowserClient(browserClient);
webServer.setSessionImporter(sessionImporter);
webServer.setAccountAuth(accountAuth);
webServer.setRedditAccountsService(redditAccounts);
webServer.setAccountRuntimeManager(accountRuntimeManager);
webServer.setConversationStateRepository(conversationStateRepo);
webServer.setConfigStore(new ConfigStore(db));
webServer.setOnConfigChange(() => logger.info('Runtime config updated via dashboard'));
@@ -337,6 +409,10 @@ export async function bootstrap(
// Start background services
if (webServer) webServer.start();
await accountRuntimeManager.startEnabled().catch((error) => {
logger.error('One or more Reddit account runtimes failed to start:', error);
});
if (accountRuntimeManager.list().length > 0) await browserClient.close();
// Prime processor with successful examples already on disk.
const existingExamples = responseChecker.getSuccessfulExamples(3);
@@ -352,6 +428,8 @@ export async function bootstrap(
db,
browserClient,
accountAuth,
redditAccountsRepo,
accountRuntimeManager,
postsRepo,
messagesRepo,
successfulMessagesRepo,
src/browser/client.ts
+16 -3
View File
@@ -88,6 +88,9 @@ export class BrowserClient {
if (!this.context) {
await this.initialize();
}
if (typeof this.context!.clearCookies === 'function') {
await this.context!.clearCookies();
}
await this.context!.addCookies(cookies);
this.sessionState = 'unknown';
}
@@ -113,7 +116,16 @@ export class BrowserClient {
/** Returns the account name from Reddit's authenticated identity endpoint. */
async getAuthenticatedUsername(): Promise<string | null> {
try {
return await this.getRedditIdentity(await this.getPage());
return (await this.getRedditIdentity(await this.getPage()))?.username ?? null;
} catch {
return null;
}
}
async getAuthenticatedIdentity(): Promise<{ username: string; accountId: string } | null> {
try {
const identity = await this.getRedditIdentity(await this.getPage());
return identity?.accountId ? { username: identity.username, accountId: identity.accountId } : null;
} catch {
return null;
}
@@ -168,7 +180,7 @@ export class BrowserClient {
return (await this.getRedditIdentity(page)) !== null;
}
private async getRedditIdentity(page: Page): Promise<string | null> {
private async getRedditIdentity(page: Page): Promise<{ username: string; accountId: string | null } | null> {
try {
return await page.evaluate(async () => {
try {
@@ -181,7 +193,8 @@ export class BrowserClient {
: data;
if (!('name' in account) || typeof account.name !== 'string') return null;
const username = account.name.trim();
return username || null;
const accountId = 'id' in account && typeof account.id === 'string' ? account.id.trim() : '';
return username ? { username, accountId: accountId || null } : null;
} catch {
return null;
}
src/database/account-repositories.ts
+375
View File
@@ -0,0 +1,375 @@
import type Database from 'better-sqlite3';
export type RedditAccountStatus = 'verified' | 'login-required' | 'disabled' | 'legacy-unassigned';
export interface RedditAccountPublicRecord {
accountId: string;
redditUsername: string;
matrixUserId: string | null;
status: RedditAccountStatus;
isActive: boolean;
createdAt: number;
updatedAt: number;
lastVerifiedAt: number | null;
lastImportedAt: number | null;
}
/** Private credential-bearing record. Never return this from a web API. */
export interface RedditAccountPrivateRecord extends RedditAccountPublicRecord {
cookiesJson: string;
matrixCredentialsJson: string;
}
export interface VerifiedRedditAccountInput {
accountId: string;
redditUsername: string;
matrixUserId: string;
cookiesJson: string;
matrixCredentialsJson: string;
now?: number;
}
type AccountRow = {
account_id: string;
reddit_username: string;
matrix_user_id: string | null;
status: RedditAccountStatus;
is_active: number;
cookies_json: string;
matrix_credentials_json: string;
created_at: number;
updated_at: number;
last_verified_at: number | null;
last_imported_at: number | null;
};
export class RedditAccountsRepository {
constructor(private readonly db: Database.Database) {}
listPublic(): RedditAccountPublicRecord[] {
const rows = this.db.prepare(`
SELECT account_id, reddit_username, matrix_user_id, status, is_active,
created_at, updated_at, last_verified_at, last_imported_at
FROM reddit_accounts
ORDER BY is_active DESC, reddit_username COLLATE NOCASE, account_id
`).all() as AccountRow[];
return rows.map(mapPublicAccount);
}
/** Private lookup for the runtime/session factory. Contains credentials. */
get(accountId: string): RedditAccountPrivateRecord | null {
const row = this.db.prepare('SELECT * FROM reddit_accounts WHERE account_id = ?')
.get(accountId) as AccountRow | undefined;
return row ? mapPrivateAccount(row) : null;
}
getActive(): RedditAccountPublicRecord[] {
const rows = this.db.prepare(`
SELECT account_id, reddit_username, matrix_user_id, status, is_active,
created_at, updated_at, last_verified_at, last_imported_at
FROM reddit_accounts
WHERE is_active = 1 AND status != 'legacy-unassigned'
ORDER BY reddit_username COLLATE NOCASE, account_id
`).all() as AccountRow[];
return rows.map(mapPublicAccount);
}
upsertVerified(input: VerifiedRedditAccountInput): RedditAccountPublicRecord {
requireText(input.accountId, 'accountId');
requireText(input.redditUsername, 'redditUsername');
requireText(input.matrixUserId, 'matrixUserId');
requireJson(input.cookiesJson, 'cookiesJson');
requireJson(input.matrixCredentialsJson, 'matrixCredentialsJson');
const now = input.now ?? Date.now();
this.db.prepare(`
INSERT INTO reddit_accounts (
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
) VALUES (
@accountId, @redditUsername, @matrixUserId, 'verified', 1,
@cookiesJson, @matrixCredentialsJson, @now, @now, @now, @now
)
ON CONFLICT(account_id) DO UPDATE SET
reddit_username = excluded.reddit_username,
matrix_user_id = excluded.matrix_user_id,
status = 'verified',
is_active = 1,
cookies_json = excluded.cookies_json,
matrix_credentials_json = excluded.matrix_credentials_json,
updated_at = excluded.updated_at,
last_verified_at = excluded.last_verified_at,
last_imported_at = excluded.last_imported_at
`).run({ ...input, now });
return publicFromPrivate(this.getRequired(input.accountId));
}
setActive(accountId: string, active = true, now: number = Date.now()): boolean {
const status = active ? 'verified' : 'disabled';
return this.db.prepare(`
UPDATE reddit_accounts
SET is_active = ?, status = ?, updated_at = ?
WHERE account_id = ? AND status != 'legacy-unassigned'
`).run(active ? 1 : 0, status, now, accountId).changes === 1;
}
markLoginRequired(accountId: string, now: number = Date.now()): boolean {
return this.db.prepare(`
UPDATE reddit_accounts
SET status = 'login-required', updated_at = ?
WHERE account_id = ? AND status != 'legacy-unassigned'
`).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`);
return account;
}
}
function mapPublicAccount(row: AccountRow): RedditAccountPublicRecord {
return {
accountId: row.account_id,
redditUsername: row.reddit_username,
matrixUserId: row.matrix_user_id,
status: row.status,
isActive: row.is_active === 1,
createdAt: row.created_at,
updatedAt: row.updated_at,
lastVerifiedAt: row.last_verified_at,
lastImportedAt: row.last_imported_at,
};
}
function mapPrivateAccount(row: AccountRow): RedditAccountPrivateRecord {
return {
...mapPublicAccount(row),
cookiesJson: row.cookies_json,
matrixCredentialsJson: row.matrix_credentials_json,
};
}
function publicFromPrivate(record: RedditAccountPrivateRecord): RedditAccountPublicRecord {
const { cookiesJson: _cookies, matrixCredentialsJson: _matrix, ...publicRecord } = record;
return publicRecord;
}
export interface AccountConversationRecord {
accountId: string;
conversationId: string;
username: string;
roomName: string | null;
payloadJson: string;
updatedAt: number;
}
export interface AccountSyncStateRecord {
accountId: string;
nextBatch: string | null;
lastFullSyncAt: number | null;
lastDeltaSyncAt: number | null;
cooldownJson: string | null;
updatedAt: number;
}
export interface AccountMediaRecord {
accountId: string;
mediaId: string;
chatUsername: string | null;
chatRoomId: string | null;
sender: string;
timestamp: number;
originalUrl: string;
fileName: string;
relativePath: string;
createdAt: number;
}
type ConversationRow = {
account_id: string;
conversation_id: string;
username: string;
room_name: string | null;
payload_json: string;
updated_at: number;
};
type SyncStateRow = {
account_id: string;
next_batch: string | null;
last_full_sync_at: number | null;
last_delta_sync_at: number | null;
cooldown_json: string | null;
updated_at: number;
};
type MediaRow = {
account_id: string;
media_id: string;
chat_username: string | null;
chat_room_id: string | null;
sender: string;
timestamp: number;
original_url: string;
file_name: string;
relative_path: string;
created_at: number;
};
/** Account-scoped DB persistence for chat cache state and downloaded-media metadata. */
export class AccountContentRepository {
constructor(private readonly db: Database.Database) {}
upsertConversation(input: Omit<AccountConversationRecord, 'updatedAt'> & { updatedAt?: number }): void {
requireText(input.accountId, 'accountId');
requireText(input.conversationId, 'conversationId');
requireText(input.username, 'username');
requireJson(input.payloadJson, 'payloadJson');
this.db.prepare(`
INSERT INTO reddit_conversations (
account_id, conversation_id, username, room_name, payload_json, updated_at
) VALUES (@accountId, @conversationId, @username, @roomName, @payloadJson, @updatedAt)
ON CONFLICT(account_id, conversation_id) DO UPDATE SET
username = excluded.username,
room_name = excluded.room_name,
payload_json = excluded.payload_json,
updated_at = excluded.updated_at
`).run({ ...input, roomName: input.roomName ?? null, updatedAt: input.updatedAt ?? Date.now() });
}
getConversation(accountId: string, conversationId: string): AccountConversationRecord | null {
const row = this.db.prepare(`
SELECT * FROM reddit_conversations WHERE account_id = ? AND conversation_id = ?
`).get(accountId, conversationId) as ConversationRow | undefined;
return row ? mapConversation(row) : null;
}
listConversations(accountId: string): AccountConversationRecord[] {
const rows = this.db.prepare(`
SELECT * FROM reddit_conversations WHERE account_id = ?
ORDER BY updated_at DESC, conversation_id
`).all(accountId) as ConversationRow[];
return rows.map(mapConversation);
}
deleteConversation(accountId: string, conversationId: string): boolean {
return this.db.prepare(`
DELETE FROM reddit_conversations WHERE account_id = ? AND conversation_id = ?
`).run(accountId, conversationId).changes === 1;
}
saveSyncState(input: Omit<AccountSyncStateRecord, 'updatedAt'> & { updatedAt?: number }): void {
if (input.cooldownJson !== null) requireJson(input.cooldownJson, 'cooldownJson');
this.db.prepare(`
INSERT INTO reddit_account_sync_state (
account_id, next_batch, last_full_sync_at, last_delta_sync_at, cooldown_json, updated_at
) VALUES (@accountId, @nextBatch, @lastFullSyncAt, @lastDeltaSyncAt, @cooldownJson, @updatedAt)
ON CONFLICT(account_id) DO UPDATE SET
next_batch = excluded.next_batch,
last_full_sync_at = excluded.last_full_sync_at,
last_delta_sync_at = excluded.last_delta_sync_at,
cooldown_json = excluded.cooldown_json,
updated_at = excluded.updated_at
`).run({ ...input, updatedAt: input.updatedAt ?? Date.now() });
}
getSyncState(accountId: string): AccountSyncStateRecord | null {
const row = this.db.prepare('SELECT * FROM reddit_account_sync_state WHERE account_id = ?')
.get(accountId) as SyncStateRow | undefined;
return row ? mapSyncState(row) : null;
}
upsertMedia(input: AccountMediaRecord): void {
requireText(input.accountId, 'accountId');
requireText(input.mediaId, 'mediaId');
this.db.prepare(`
INSERT INTO reddit_media (
account_id, media_id, chat_username, chat_room_id, sender, timestamp,
original_url, file_name, relative_path, created_at
) VALUES (
@accountId, @mediaId, @chatUsername, @chatRoomId, @sender, @timestamp,
@originalUrl, @fileName, @relativePath, @createdAt
)
ON CONFLICT(account_id, media_id) DO UPDATE SET
chat_username = excluded.chat_username,
chat_room_id = excluded.chat_room_id,
sender = excluded.sender,
timestamp = excluded.timestamp,
original_url = excluded.original_url,
file_name = excluded.file_name,
relative_path = excluded.relative_path,
created_at = excluded.created_at
`).run(input);
}
getMedia(accountId: string, mediaId: string): AccountMediaRecord | null {
const row = this.db.prepare(`
SELECT * FROM reddit_media WHERE account_id = ? AND media_id = ?
`).get(accountId, mediaId) as MediaRow | undefined;
return row ? mapMedia(row) : null;
}
listMedia(accountId: string): AccountMediaRecord[] {
const rows = this.db.prepare(`
SELECT * FROM reddit_media WHERE account_id = ?
ORDER BY timestamp DESC, media_id
`).all(accountId) as MediaRow[];
return rows.map(mapMedia);
}
deleteMedia(accountId: string, mediaId: string): boolean {
return this.db.prepare(`
DELETE FROM reddit_media WHERE account_id = ? AND media_id = ?
`).run(accountId, mediaId).changes === 1;
}
}
function mapConversation(row: ConversationRow): AccountConversationRecord {
return {
accountId: row.account_id,
conversationId: row.conversation_id,
username: row.username,
roomName: row.room_name,
payloadJson: row.payload_json,
updatedAt: row.updated_at,
};
}
function mapSyncState(row: SyncStateRow): AccountSyncStateRecord {
return {
accountId: row.account_id,
nextBatch: row.next_batch,
lastFullSyncAt: row.last_full_sync_at,
lastDeltaSyncAt: row.last_delta_sync_at,
cooldownJson: row.cooldown_json,
updatedAt: row.updated_at,
};
}
function mapMedia(row: MediaRow): AccountMediaRecord {
return {
accountId: row.account_id,
mediaId: row.media_id,
chatUsername: row.chat_username,
chatRoomId: row.chat_room_id,
sender: row.sender,
timestamp: row.timestamp,
originalUrl: row.original_url,
fileName: row.file_name,
relativePath: row.relative_path,
createdAt: row.created_at,
};
}
function requireText(value: string, name: string): void {
if (!value.trim()) throw new Error(`${name} must not be empty`);
}
function requireJson(value: string, name: string): void {
try {
JSON.parse(value);
} catch {
throw new Error(`${name} must be valid JSON`);
}
}
src/database/client.ts
+264 -4
View File
@@ -3,6 +3,8 @@ import type { Config } from '../config/index.js';
import path from 'path';
import fs from 'fs';
export const LEGACY_UNASSIGNED_ACCOUNT_ID = '__legacy_unassigned__';
export function createDatabase(config: Pick<Config, 'database'>): Database.Database {
const dbPath = config.database.path === ':memory:'
? ':memory:'
@@ -198,6 +200,7 @@ export function runMigrations(db: Database.Database): void {
CREATE INDEX IF NOT EXISTS idx_sent_messages_image ON sent_messages(got_image);
`);
},
() => migrateAccountOwnership(db),
];
let currentVersion = db.pragma('user_version', { simple: true }) as number;
@@ -207,14 +210,271 @@ export function runMigrations(db: Database.Database): void {
for (let index = currentVersion; index < migrations.length; index += 1) {
const version = index + 1;
db.transaction(() => {
migrations[index]();
db.pragma(`user_version = ${version}`);
})();
// SQLite cannot rebuild a referenced table while FK enforcement is on.
// Migration 11 replaces the account-produced table graph atomically, then
// validates it before foreign-key enforcement is restored.
const rebuildsForeignKeyGraph = version === 11;
if (rebuildsForeignKeyGraph) db.pragma('foreign_keys = OFF');
try {
db.transaction(() => {
migrations[index]();
if (rebuildsForeignKeyGraph) {
const violations = db.pragma('foreign_key_check') as unknown[];
if (violations.length > 0) {
throw new Error(`Account ownership migration left ${violations.length} foreign-key violation(s)`);
}
}
db.pragma(`user_version = ${version}`);
})();
} finally {
if (rebuildsForeignKeyGraph) db.pragma('foreign_keys = ON');
}
currentVersion = version;
}
}
function migrateAccountOwnership(db: Database.Database): void {
const legacy = LEGACY_UNASSIGNED_ACCOUNT_ID.replaceAll("'", "''");
db.exec(`
CREATE TABLE reddit_accounts (
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', 'legacy-unassigned')),
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
);
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);
INSERT INTO reddit_accounts (
account_id, reddit_username, matrix_user_id, status, is_active,
cookies_json, matrix_credentials_json, created_at, updated_at
) VALUES (
'${legacy}', 'Legacy / unassigned', NULL, 'legacy-unassigned', 0,
'[]', '{}', 0, 0
);
CREATE TABLE processed_posts_account_scoped (
account_id TEXT NOT NULL DEFAULT '${legacy}',
id TEXT NOT NULL,
subreddit TEXT NOT NULL,
author TEXT NOT NULL,
title TEXT NOT NULL,
image_url TEXT,
image_analysis TEXT,
generated_message TEXT,
processed_at INTEGER NOT NULL,
message_sent INTEGER DEFAULT 0,
error TEXT,
post_url TEXT,
PRIMARY KEY (account_id, id),
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id)
);
INSERT INTO processed_posts_account_scoped (
account_id, id, subreddit, author, title, image_url, image_analysis,
generated_message, processed_at, message_sent, error, post_url
) SELECT '${legacy}', id, subreddit, author, title, image_url, image_analysis,
generated_message, processed_at, message_sent, error, post_url
FROM processed_posts;
CREATE TABLE sent_messages_account_scoped (
id INTEGER PRIMARY KEY AUTOINCREMENT,
account_id TEXT NOT NULL DEFAULT '${legacy}',
post_id TEXT,
recipient TEXT NOT NULL,
subject TEXT NOT NULL,
message TEXT NOT NULL,
sent_at INTEGER NOT NULL,
got_response INTEGER DEFAULT 0,
got_image INTEGER NOT NULL DEFAULT 0,
response_checked_at INTEGER,
strategy_id TEXT,
chat_room_id TEXT,
UNIQUE (account_id, id),
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id),
FOREIGN KEY (account_id, post_id) REFERENCES processed_posts_account_scoped(account_id, id)
);
INSERT INTO sent_messages_account_scoped (
id, account_id, post_id, recipient, subject, message, sent_at,
got_response, got_image, response_checked_at, strategy_id, chat_room_id
) SELECT id, '${legacy}', post_id, recipient, subject, message, sent_at,
got_response, got_image, response_checked_at, strategy_id, chat_room_id
FROM sent_messages;
CREATE TABLE successful_messages_account_scoped (
id INTEGER PRIMARY KEY AUTOINCREMENT,
account_id TEXT NOT NULL DEFAULT '${legacy}',
sent_message_id INTEGER,
original_message TEXT NOT NULL,
image_analysis TEXT,
post_title TEXT,
response_snippet TEXT,
created_at INTEGER NOT NULL,
subreddit TEXT,
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id),
FOREIGN KEY (account_id, sent_message_id) REFERENCES sent_messages_account_scoped(account_id, id)
);
INSERT INTO successful_messages_account_scoped (
id, account_id, sent_message_id, original_message, image_analysis,
post_title, response_snippet, created_at, subreddit
) SELECT id, '${legacy}', sent_message_id, original_message, image_analysis,
post_title, response_snippet, created_at, subreddit
FROM successful_messages;
CREATE TABLE activity_log_account_scoped (
id INTEGER PRIMARY KEY AUTOINCREMENT,
account_id TEXT NOT NULL DEFAULT '${legacy}',
timestamp INTEGER NOT NULL,
type TEXT NOT NULL,
message TEXT NOT NULL,
recipient TEXT,
image_url TEXT,
post_title TEXT,
post_url TEXT,
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id)
);
INSERT INTO activity_log_account_scoped (
id, account_id, timestamp, type, message, recipient, image_url, post_title, post_url
) SELECT id, '${legacy}', timestamp, type, message, recipient, image_url, post_title, post_url
FROM activity_log;
CREATE TABLE delivery_outbox_account_scoped (
id INTEGER PRIMARY KEY AUTOINCREMENT,
account_id TEXT NOT NULL DEFAULT '${legacy}',
post_id TEXT NOT NULL,
recipient TEXT NOT NULL,
subject TEXT NOT NULL,
message TEXT NOT NULL,
strategy_id TEXT,
idempotency_key TEXT NOT NULL,
status TEXT NOT NULL CHECK(status IN ('discovered', 'prepared', 'sending', 'sent', 'failed')),
created_at INTEGER NOT NULL,
updated_at INTEGER NOT NULL,
lease_owner TEXT,
lease_expires_at INTEGER,
attempt_count INTEGER NOT NULL DEFAULT 0,
last_error TEXT,
sent_message_id INTEGER,
UNIQUE (account_id, post_id),
UNIQUE (account_id, idempotency_key),
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id),
FOREIGN KEY (account_id, post_id) REFERENCES processed_posts_account_scoped(account_id, id),
FOREIGN KEY (account_id, sent_message_id) REFERENCES sent_messages_account_scoped(account_id, id)
);
INSERT INTO delivery_outbox_account_scoped (
id, account_id, post_id, recipient, subject, message, strategy_id,
idempotency_key, status, created_at, updated_at, lease_owner,
lease_expires_at, attempt_count, last_error, sent_message_id
) SELECT id, '${legacy}', post_id, recipient, subject, message, strategy_id,
idempotency_key, status, created_at, updated_at, lease_owner,
lease_expires_at, attempt_count, last_error, sent_message_id
FROM delivery_outbox;
CREATE TABLE conversation_states_account_scoped (
account_id TEXT NOT NULL DEFAULT '${legacy}',
conversation_id TEXT NOT NULL,
state_json TEXT NOT NULL,
updated_at INTEGER NOT NULL,
PRIMARY KEY (account_id, conversation_id),
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id)
);
INSERT INTO conversation_states_account_scoped (account_id, conversation_id, state_json, updated_at)
SELECT '${legacy}', conversation_id, state_json, updated_at FROM conversation_states;
CREATE TABLE conversation_outcomes_account_scoped (
account_id TEXT NOT NULL DEFAULT '${legacy}',
sent_message_id INTEGER NOT NULL,
outcome TEXT NOT NULL CHECK(outcome IN ('engaged', 'declined', 'boundary', 'upsell', 'unknown')),
response_snippet TEXT,
recorded_at INTEGER NOT NULL,
PRIMARY KEY (account_id, sent_message_id),
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id),
FOREIGN KEY (account_id, sent_message_id) REFERENCES sent_messages_account_scoped(account_id, id)
);
INSERT INTO conversation_outcomes_account_scoped (
account_id, sent_message_id, outcome, response_snippet, recorded_at
) SELECT '${legacy}', sent_message_id, outcome, response_snippet, recorded_at
FROM conversation_outcomes;
DROP TABLE conversation_outcomes;
DROP TABLE conversation_states;
DROP TABLE delivery_outbox;
DROP TABLE successful_messages;
DROP TABLE sent_messages;
DROP TABLE processed_posts;
DROP TABLE activity_log;
ALTER TABLE processed_posts_account_scoped RENAME TO processed_posts;
ALTER TABLE sent_messages_account_scoped RENAME TO sent_messages;
ALTER TABLE successful_messages_account_scoped RENAME TO successful_messages;
ALTER TABLE activity_log_account_scoped RENAME TO activity_log;
ALTER TABLE delivery_outbox_account_scoped RENAME TO delivery_outbox;
ALTER TABLE conversation_states_account_scoped RENAME TO conversation_states;
ALTER TABLE conversation_outcomes_account_scoped RENAME TO conversation_outcomes;
CREATE INDEX idx_posts_subreddit ON processed_posts(account_id, subreddit);
CREATE INDEX idx_posts_processed_at ON processed_posts(account_id, processed_at);
CREATE INDEX idx_posts_message_sent ON processed_posts(account_id, message_sent);
CREATE INDEX idx_posts_author_history
ON processed_posts(author COLLATE NOCASE, processed_at DESC, id DESC);
CREATE INDEX idx_sent_messages_response ON sent_messages(account_id, got_response);
CREATE INDEX idx_sent_messages_image ON sent_messages(account_id, got_image);
CREATE INDEX idx_sent_messages_recipient ON sent_messages(account_id, recipient);
CREATE INDEX idx_sent_messages_chat_room ON sent_messages(account_id, chat_room_id);
CREATE INDEX idx_activity_log_timestamp ON activity_log(account_id, timestamp);
CREATE INDEX idx_delivery_outbox_claim ON delivery_outbox(account_id, status, lease_expires_at, created_at);
CREATE TABLE reddit_conversations (
account_id TEXT NOT NULL,
conversation_id TEXT NOT NULL,
username TEXT NOT NULL,
room_name TEXT,
payload_json TEXT NOT NULL,
updated_at INTEGER NOT NULL,
PRIMARY KEY (account_id, conversation_id),
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id) ON DELETE CASCADE
);
CREATE INDEX idx_reddit_conversations_username
ON reddit_conversations(account_id, username COLLATE NOCASE);
CREATE INDEX idx_reddit_conversations_updated
ON reddit_conversations(account_id, updated_at DESC);
CREATE TABLE reddit_account_sync_state (
account_id TEXT PRIMARY KEY,
next_batch TEXT,
last_full_sync_at INTEGER,
last_delta_sync_at INTEGER,
cooldown_json TEXT,
updated_at INTEGER NOT NULL,
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id) ON DELETE CASCADE
);
CREATE TABLE reddit_media (
account_id TEXT NOT NULL,
media_id TEXT NOT NULL,
chat_username TEXT,
chat_room_id TEXT,
sender TEXT NOT NULL,
timestamp INTEGER NOT NULL,
original_url TEXT NOT NULL,
file_name TEXT NOT NULL,
relative_path TEXT NOT NULL,
created_at INTEGER NOT NULL,
PRIMARY KEY (account_id, media_id),
UNIQUE (account_id, relative_path),
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id) ON DELETE CASCADE
);
CREATE INDEX idx_reddit_media_timestamp ON reddit_media(account_id, timestamp DESC);
CREATE INDEX idx_reddit_media_room ON reddit_media(account_id, chat_room_id);
`);
}
function addColumnIfNotExists(db: Database.Database, table: string, column: string, definition: string): void {
const columns = db.prepare(`PRAGMA table_info(${table})`).all() as { name: string }[];
const exists = columns.some((c) => c.name === column);
src/database/legacy-content-import.ts
+95
View File
@@ -0,0 +1,95 @@
import fs from 'node:fs';
import path from 'node:path';
import type Database from 'better-sqlite3';
import { z } from 'zod';
import { LEGACY_UNASSIGNED_ACCOUNT_ID } from './client.js';
import { AccountContentRepository } from './account-repositories.js';
const ChatCacheSchema = z.object({
chats: z.array(z.object({
username: z.string().min(1),
roomName: z.string(),
roomId: z.string().optional(),
}).passthrough()),
nextBatch: z.string().nullable().optional(),
lastFullSyncAt: z.number().nullable().optional(),
lastDeltaSyncAt: z.number().nullable().optional(),
cooldown: z.unknown().nullable().optional(),
});
const MediaIndexSchema = z.object({
items: z.array(z.object({
id: z.string().min(1),
chatUsername: z.string().nullable(),
chatRoomId: z.string().nullable(),
sender: z.string(),
timestamp: z.number(),
originalUrl: z.string(),
fileName: z.string(),
relativePath: z.string(),
createdAt: z.number(),
})),
});
export interface LegacyContentImportResult { conversations: number; media: number; skipped: boolean }
/** Imports pre-account JSON indexes once, preserving them as explicitly unassigned content. */
export function importLegacyAccountContent(
db: Database.Database,
dataDirectory = path.join(process.cwd(), 'data'),
): LegacyContentImportResult {
db.exec(`CREATE TABLE IF NOT EXISTS legacy_content_imports (
import_key TEXT PRIMARY KEY,
imported_at INTEGER NOT NULL
)`);
const importKey = 'account-ownership-json-v1';
if (db.prepare('SELECT 1 FROM legacy_content_imports WHERE import_key = ?').get(importKey)) {
return { conversations: 0, media: 0, skipped: true };
}
const chatPath = path.join(dataDirectory, 'chat-cache.json');
const mediaPath = path.join(dataDirectory, 'media', 'index.json');
const hasChat = fs.existsSync(chatPath);
const hasMedia = fs.existsSync(mediaPath);
if (!hasChat && !hasMedia) return { conversations: 0, media: 0, skipped: false };
const chat = hasChat ? ChatCacheSchema.parse(JSON.parse(fs.readFileSync(chatPath, 'utf8'))) : null;
const media = hasMedia ? MediaIndexSchema.parse(JSON.parse(fs.readFileSync(mediaPath, 'utf8'))) : null;
const content = new AccountContentRepository(db);
db.transaction(() => {
for (const conversation of chat?.chats ?? []) {
content.upsertConversation({
accountId: LEGACY_UNASSIGNED_ACCOUNT_ID,
conversationId: conversation.roomId ?? conversation.username,
username: conversation.username,
roomName: conversation.roomName,
payloadJson: JSON.stringify(conversation),
});
}
if (chat) {
content.saveSyncState({
accountId: LEGACY_UNASSIGNED_ACCOUNT_ID,
nextBatch: chat.nextBatch ?? null,
lastFullSyncAt: chat.lastFullSyncAt ?? null,
lastDeltaSyncAt: chat.lastDeltaSyncAt ?? null,
cooldownJson: chat.cooldown == null ? null : JSON.stringify(chat.cooldown),
});
}
for (const item of media?.items ?? []) {
content.upsertMedia({
accountId: LEGACY_UNASSIGNED_ACCOUNT_ID,
mediaId: item.id,
chatUsername: item.chatUsername,
chatRoomId: item.chatRoomId,
sender: item.sender,
timestamp: item.timestamp,
originalUrl: item.originalUrl,
fileName: item.fileName,
relativePath: item.relativePath,
createdAt: item.createdAt,
});
}
db.prepare('INSERT INTO legacy_content_imports (import_key, imported_at) VALUES (?, ?)').run(importKey, Date.now());
})();
return { conversations: chat?.chats.length ?? 0, media: media?.items.length ?? 0, skipped: false };
}
src/database/repositories.ts
+144 -101
View File
@@ -5,6 +5,7 @@ import {
type ConversationState,
} from '../services/conversation-state.js';
import type { ConversationOutcome } from '../services/conversation-outcome.js';
import { LEGACY_UNASSIGNED_ACCOUNT_ID } from './client.js';
export interface ProcessedPostRecord {
id: string;
@@ -114,7 +115,10 @@ export interface DeliveryOutboxRecord {
* lease owner. A crashed worker's lease can safely be recovered on startup.
*/
export class DeliveryOutboxRepository {
constructor(private readonly db: Database.Database) {}
constructor(
private readonly db: Database.Database,
private readonly accountId: string = LEGACY_UNASSIGNED_ACCOUNT_ID,
) {}
discover(input: {
postId: string;
@@ -128,14 +132,14 @@ export class DeliveryOutboxRepository {
const now = input.now ?? Date.now();
const insert = this.db.prepare(`
INSERT INTO delivery_outbox
(post_id, recipient, subject, message, strategy_id, idempotency_key, status, created_at, updated_at)
VALUES (@postId, @recipient, @subject, @message, @strategyId, @idempotencyKey, 'discovered', @now, @now)
ON CONFLICT(idempotency_key) DO NOTHING
(account_id, post_id, recipient, subject, message, strategy_id, idempotency_key, status, created_at, updated_at)
VALUES (@accountId, @postId, @recipient, @subject, @message, @strategyId, @idempotencyKey, 'discovered', @now, @now)
ON CONFLICT(account_id, idempotency_key) DO NOTHING
`);
const select = this.db.prepare('SELECT * FROM delivery_outbox WHERE idempotency_key = ?');
const select = this.db.prepare('SELECT * FROM delivery_outbox WHERE account_id = ? AND idempotency_key = ?');
const transaction = this.db.transaction(() => {
insert.run({ ...input, strategyId: input.strategyId ?? null, now });
const row = select.get(input.idempotencyKey) as Record<string, unknown> | undefined;
insert.run({ ...input, accountId: this.accountId, strategyId: input.strategyId ?? null, now });
const row = select.get(this.accountId, input.idempotencyKey) as Record<string, unknown> | undefined;
if (!row) throw new Error('Outbox delivery was not persisted');
return this.mapRow(row);
});
@@ -143,19 +147,21 @@ export class DeliveryOutboxRepository {
}
get(id: number): DeliveryOutboxRecord | null {
const row = this.db.prepare('SELECT * FROM delivery_outbox WHERE id = ?').get(id) as Record<string, unknown> | undefined;
const row = this.db.prepare('SELECT * FROM delivery_outbox WHERE account_id = ? AND id = ?')
.get(this.accountId, id) as Record<string, unknown> | undefined;
return row ? this.mapRow(row) : null;
}
count(): number {
return (this.db.prepare('SELECT COUNT(*) AS count FROM delivery_outbox').get() as { count: number }).count;
return (this.db.prepare('SELECT COUNT(*) AS count FROM delivery_outbox WHERE account_id = ?')
.get(this.accountId) as { count: number }).count;
}
markPrepared(id: number, now: number = Date.now()): boolean {
return this.db.prepare(`
UPDATE delivery_outbox SET status = 'prepared', updated_at = ?, last_error = NULL
WHERE id = ? AND status IN ('discovered', 'failed')
`).run(now, id).changes === 1;
WHERE account_id = ? AND id = ? AND status IN ('discovered', 'failed')
`).run(now, this.accountId, id).changes === 1;
}
claimNext(workerId: string, leaseMs: number, now: number = Date.now()): DeliveryOutboxRecord | null {
@@ -164,16 +170,16 @@ export class DeliveryOutboxRepository {
const transaction = this.db.transaction(() => {
const row = this.db.prepare(`
SELECT id FROM delivery_outbox
WHERE status = 'prepared'
WHERE account_id = ? AND status = 'prepared'
ORDER BY created_at ASC, id ASC
LIMIT 1
`).get() as { id: number } | undefined;
`).get(this.accountId) as { id: number } | undefined;
if (!row) return null;
const changed = this.db.prepare(`
UPDATE delivery_outbox
SET status = 'sending', lease_owner = ?, lease_expires_at = ?, updated_at = ?, attempt_count = attempt_count + 1
WHERE id = ? AND status = 'prepared'
`).run(workerId, now + leaseMs, now, row.id).changes;
WHERE account_id = ? AND id = ? AND status = 'prepared'
`).run(workerId, now + leaseMs, now, this.accountId, row.id).changes;
if (changed !== 1) return null;
return this.get(row.id);
});
@@ -187,8 +193,8 @@ export class DeliveryOutboxRepository {
const changed = this.db.prepare(`
UPDATE delivery_outbox
SET status = 'sending', lease_owner = ?, lease_expires_at = ?, updated_at = ?, attempt_count = attempt_count + 1
WHERE id = ? AND status = 'prepared'
`).run(workerId, now + leaseMs, now, id).changes;
WHERE account_id = ? AND id = ? AND status = 'prepared'
`).run(workerId, now + leaseMs, now, this.accountId, id).changes;
return changed === 1 ? this.get(id) : null;
}
@@ -196,24 +202,24 @@ export class DeliveryOutboxRepository {
return this.db.prepare(`
UPDATE delivery_outbox
SET status = 'prepared', lease_owner = NULL, lease_expires_at = NULL, updated_at = ?
WHERE status = 'sending' AND lease_expires_at <= ?
`).run(now, now).changes;
WHERE account_id = ? AND status = 'sending' AND lease_expires_at <= ?
`).run(now, this.accountId, now).changes;
}
markSent(id: number, workerId: string, sentMessageId: number, now: number = Date.now()): boolean {
return this.db.prepare(`
UPDATE delivery_outbox
SET status = 'sent', sent_message_id = ?, lease_owner = NULL, lease_expires_at = NULL, updated_at = ?, last_error = NULL
WHERE id = ? AND status = 'sending' AND lease_owner = ?
`).run(sentMessageId, now, id, workerId).changes === 1;
WHERE account_id = ? AND id = ? AND status = 'sending' AND lease_owner = ?
`).run(sentMessageId, now, this.accountId, id, workerId).changes === 1;
}
markFailed(id: number, workerId: string, error: string, now: number = Date.now()): boolean {
return this.db.prepare(`
UPDATE delivery_outbox
SET status = 'failed', last_error = ?, lease_owner = NULL, lease_expires_at = NULL, updated_at = ?
WHERE id = ? AND status = 'sending' AND lease_owner = ?
`).run(error, now, id, workerId).changes === 1;
WHERE account_id = ? AND id = ? AND status = 'sending' AND lease_owner = ?
`).run(error, now, this.accountId, id, workerId).changes === 1;
}
private mapRow(row: Record<string, unknown>): DeliveryOutboxRecord {
@@ -243,26 +249,34 @@ export class ActivityLogRepository {
private pruneStmt: Database.Statement;
private countSentSkippedStmt: Database.Statement;
constructor(private db: Database.Database) {
constructor(
private db: Database.Database,
private readonly accountId: string = LEGACY_UNASSIGNED_ACCOUNT_ID,
) {
this.insertStmt = db.prepare(`
INSERT INTO activity_log (timestamp, type, message, recipient, image_url, post_title, post_url)
VALUES (?, ?, ?, ?, ?, ?, ?)
INSERT INTO activity_log (account_id, timestamp, type, message, recipient, image_url, post_title, post_url)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
`);
this.getRecentStmt = db.prepare(`
SELECT timestamp, type, message, recipient, image_url, post_title, post_url
FROM activity_log
WHERE account_id = ?
ORDER BY id DESC
LIMIT ?
`);
this.pruneStmt = db.prepare(`
DELETE FROM activity_log
WHERE id NOT IN (SELECT id FROM activity_log ORDER BY id DESC LIMIT ?)
WHERE account_id = ?
AND id NOT IN (
SELECT id FROM activity_log WHERE account_id = ? ORDER BY id DESC LIMIT ?
)
`);
this.countSentSkippedStmt = db.prepare(`
SELECT COUNT(*) AS count FROM activity_log WHERE type IN ('sent', 'skipped')
SELECT COUNT(*) AS count FROM activity_log
WHERE account_id = ? AND type IN ('sent', 'skipped')
`);
}
@@ -276,6 +290,7 @@ export class ActivityLogRepository {
postUrl?: string;
}): void {
this.insertStmt.run(
this.accountId,
item.timestamp,
item.type,
item.message,
@@ -287,7 +302,7 @@ export class ActivityLogRepository {
}
getRecent(limit: number): ActivityLogRecord[] {
const rows = this.getRecentStmt.all(limit) as Record<string, unknown>[];
const rows = this.getRecentStmt.all(this.accountId, limit) as Record<string, unknown>[];
return rows.map((row) => ({
timestamp: row.timestamp as number,
type: row.type as string,
@@ -300,11 +315,11 @@ export class ActivityLogRepository {
}
prune(keep: number): void {
this.pruneStmt.run(keep);
this.pruneStmt.run(this.accountId, this.accountId, keep);
}
countSentSkipped(): number {
const row = this.countSentSkippedStmt.get() as { count: number };
const row = this.countSentSkippedStmt.get(this.accountId) as { count: number };
return row.count;
}
}
@@ -318,33 +333,38 @@ export class PostsRepository {
private getByIdStmt: Database.Statement;
private getByAuthorStmt: Database.Statement;
constructor(private db: Database.Database) {
constructor(
private db: Database.Database,
private readonly accountId: string = LEGACY_UNASSIGNED_ACCOUNT_ID,
) {
this.insertStmt = db.prepare(`
INSERT INTO processed_posts (id, subreddit, author, title, post_url, image_url, processed_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
INSERT INTO processed_posts (account_id, id, subreddit, author, title, post_url, image_url, processed_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
`);
this.existsStmt = db.prepare(`SELECT 1 FROM processed_posts WHERE id = ?`);
this.existsStmt = db.prepare(`SELECT 1 FROM processed_posts WHERE account_id = ? AND id = ?`);
this.markSentStmt = db.prepare(`UPDATE processed_posts SET message_sent = 1 WHERE id = ?`);
this.markSentStmt = db.prepare(`UPDATE processed_posts SET message_sent = 1 WHERE account_id = ? AND id = ?`);
this.updateAnalysisStmt = db.prepare(`
UPDATE processed_posts
SET image_analysis = ?, generated_message = ?
WHERE id = ?
WHERE account_id = ? AND id = ?
`);
this.setErrorStmt = db.prepare(`UPDATE processed_posts SET error = ? WHERE id = ?`);
this.setErrorStmt = db.prepare(`UPDATE processed_posts SET error = ? WHERE account_id = ? AND id = ?`);
this.getByIdStmt = db.prepare(`SELECT * FROM processed_posts WHERE id = ?`);
this.getByIdStmt = db.prepare(`SELECT * FROM processed_posts WHERE account_id = ? AND id = ?`);
this.getByAuthorStmt = db.prepare(`
SELECT * FROM processed_posts WHERE author = ? COLLATE NOCASE ORDER BY processed_at DESC LIMIT 10
SELECT * FROM processed_posts
WHERE account_id = ? AND author = ? COLLATE NOCASE
ORDER BY processed_at DESC, id DESC LIMIT 10
`);
}
isProcessed(postId: string): boolean {
return this.existsStmt.get(postId) !== undefined;
return this.existsStmt.get(this.accountId, postId) !== undefined;
}
insert(post: {
@@ -357,6 +377,7 @@ export class PostsRepository {
processedAt: number;
}): void {
this.insertStmt.run(
this.accountId,
post.id,
post.subreddit,
post.author,
@@ -368,19 +389,19 @@ export class PostsRepository {
}
updateAnalysis(postId: string, analysis: string, message: string): void {
this.updateAnalysisStmt.run(analysis, message, postId);
this.updateAnalysisStmt.run(analysis, message, this.accountId, postId);
}
markMessageSent(postId: string): void {
this.markSentStmt.run(postId);
this.markSentStmt.run(this.accountId, postId);
}
setError(postId: string, error: string): void {
this.setErrorStmt.run(error, postId);
this.setErrorStmt.run(error, this.accountId, postId);
}
getById(postId: string): ProcessedPostRecord | undefined {
const row = this.getByIdStmt.get(postId) as Record<string, unknown> | undefined;
const row = this.getByIdStmt.get(this.accountId, postId) as Record<string, unknown> | undefined;
if (!row) return undefined;
return {
@@ -399,7 +420,7 @@ export class PostsRepository {
}
getByAuthor(author: string): ProcessedPostRecord[] {
const rows = this.getByAuthorStmt.all(author) as Record<string, unknown>[];
const rows = this.getByAuthorStmt.all(this.accountId, author) as Record<string, unknown>[];
return rows.map((row) => ({
id: row.id as string,
subreddit: row.subreddit as string,
@@ -429,35 +450,38 @@ export class MessagesRepository {
private markCheckedStmt: Database.Statement;
private hasSentToStmt: Database.Statement;
constructor(private db: Database.Database) {
constructor(
private db: Database.Database,
private readonly accountId: string = LEGACY_UNASSIGNED_ACCOUNT_ID,
) {
this.insertStmt = db.prepare(`
INSERT INTO sent_messages (post_id, recipient, subject, message, sent_at, strategy_id, chat_room_id)
VALUES (?, ?, ?, ?, ?, ?, ?)
INSERT INTO sent_messages (account_id, post_id, recipient, subject, message, sent_at, strategy_id, chat_room_id)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
`);
this.insertChatStmt = db.prepare(`
INSERT INTO sent_messages (post_id, recipient, subject, message, sent_at, chat_room_id)
VALUES (NULL, ?, ?, ?, ?, ?)
INSERT INTO sent_messages (account_id, post_id, recipient, subject, message, sent_at, chat_room_id)
VALUES (?, NULL, ?, ?, ?, ?, ?)
`);
this.getByChatRoomIdStmt = db.prepare(`
SELECT sm.*, pp.subreddit, pp.title AS post_title, pp.post_url, pp.image_url AS post_image_url,
pp.image_analysis AS post_image_analysis
FROM sent_messages sm
LEFT JOIN processed_posts pp ON sm.post_id = pp.id
WHERE sm.chat_room_id = ?
LEFT JOIN processed_posts pp ON sm.account_id = pp.account_id AND sm.post_id = pp.id
WHERE sm.account_id = ? AND sm.chat_room_id = ?
ORDER BY sm.post_id IS NULL ASC, sm.sent_at DESC
LIMIT 1
`);
this.getByPostIdStmt = db.prepare(`SELECT * FROM sent_messages WHERE post_id = ?`);
this.getByPostIdStmt = db.prepare(`SELECT * FROM sent_messages WHERE account_id = ? AND post_id = ?`);
this.getLatestByRecipientStmt = db.prepare(`
SELECT sm.*, pp.subreddit, pp.title AS post_title, pp.post_url, pp.image_url AS post_image_url,
pp.image_analysis AS post_image_analysis
FROM sent_messages sm
LEFT JOIN processed_posts pp ON sm.post_id = pp.id
WHERE sm.recipient = ? COLLATE NOCASE
LEFT JOIN processed_posts pp ON sm.account_id = pp.account_id AND sm.post_id = pp.id
WHERE sm.account_id = ? AND sm.recipient = ? COLLATE NOCASE
ORDER BY sm.sent_at DESC
LIMIT 1
`);
@@ -466,14 +490,15 @@ export class MessagesRepository {
SELECT sm.*, pp.subreddit, pp.title AS post_title, pp.post_url, pp.image_url AS post_image_url,
pp.image_analysis AS post_image_analysis
FROM sent_messages sm
LEFT JOIN processed_posts pp ON sm.post_id = pp.id
LEFT JOIN processed_posts pp ON sm.account_id = pp.account_id AND sm.post_id = pp.id
WHERE sm.account_id = ?
ORDER BY sm.sent_at DESC
LIMIT ?
`);
this.getUncheckedStmt = db.prepare(`
SELECT * FROM sent_messages
WHERE got_response = 0
WHERE account_id = ? AND got_response = 0
AND (response_checked_at IS NULL OR response_checked_at < ?)
AND sent_at < ?
ORDER BY sent_at ASC
@@ -481,7 +506,7 @@ export class MessagesRepository {
`);
this.markResponseStmt = db.prepare(`
UPDATE sent_messages SET got_response = 1, response_checked_at = ? WHERE id = ?
UPDATE sent_messages SET got_response = 1, response_checked_at = ? WHERE account_id = ? AND id = ?
`);
this.markImageStmt = db.prepare(`
@@ -491,11 +516,11 @@ export class MessagesRepository {
`);
this.markCheckedStmt = db.prepare(`
UPDATE sent_messages SET response_checked_at = ? WHERE id = ?
UPDATE sent_messages SET response_checked_at = ? WHERE account_id = ? AND id = ?
`);
this.hasSentToStmt = db.prepare(`
SELECT 1 FROM sent_messages WHERE recipient = ? LIMIT 1
SELECT 1 FROM sent_messages WHERE account_id = ? AND recipient = ? LIMIT 1
`);
}
@@ -509,6 +534,7 @@ export class MessagesRepository {
chatRoomId?: string | null;
}): number {
const result = this.insertStmt.run(
this.accountId,
message.postId,
message.recipient,
message.subject,
@@ -528,6 +554,7 @@ export class MessagesRepository {
chatRoomId: string;
}): number {
const result = this.insertChatStmt.run(
this.accountId,
message.recipient,
message.subject,
message.message,
@@ -538,33 +565,34 @@ export class MessagesRepository {
}
getByPostId(postId: string): SentMessageRecord | undefined {
const row = this.getByPostIdStmt.get(postId) as Record<string, unknown> | undefined;
const row = this.getByPostIdStmt.get(this.accountId, postId) as Record<string, unknown> | undefined;
if (!row) return undefined;
return this.mapRow(row);
}
getByChatRoomId(chatRoomId: string): SentMessageRecord | undefined {
const row = this.getByChatRoomIdStmt.get(chatRoomId) as Record<string, unknown> | undefined;
const row = this.getByChatRoomIdStmt.get(this.accountId, chatRoomId) as Record<string, unknown> | undefined;
if (!row) return undefined;
return this.mapRow(row);
}
getLatestByRecipient(recipient: string): SentMessageRecord | undefined {
const row = this.getLatestByRecipientStmt.get(recipient) as Record<string, unknown> | undefined;
const row = this.getLatestByRecipientStmt.get(this.accountId, recipient) as Record<string, unknown> | undefined;
if (!row) return undefined;
return this.mapRow(row);
}
getRecent(limit: number = 10): SentMessageRecord[] {
const rows = this.getRecentStmt.all(limit) as Record<string, unknown>[];
const rows = this.getRecentStmt.all(this.accountId, limit) as Record<string, unknown>[];
return rows.map((row) => this.mapRow(row));
}
/** Successful-send history adapter for durable rolling rate limits. */
getSentTimestampsSince(since: number): number[] {
const rows = this.db.prepare(`
SELECT sent_at FROM sent_messages WHERE sent_at > ? ORDER BY sent_at ASC
`).all(since) as Array<{ sent_at: number }>;
SELECT sent_at FROM sent_messages
WHERE account_id = ? AND sent_at > ? ORDER BY sent_at ASC
`).all(this.accountId, since) as Array<{ sent_at: number }>;
return rows.map((row) => row.sent_at);
}
@@ -572,12 +600,12 @@ export class MessagesRepository {
const now = Date.now();
const minSentAt = now - minAgeMs;
const recheckBefore = now - recheckAfterMs;
const rows = this.getUncheckedStmt.all(recheckBefore, minSentAt, limit) as Record<string, unknown>[];
const rows = this.getUncheckedStmt.all(this.accountId, recheckBefore, minSentAt, limit) as Record<string, unknown>[];
return rows.map((row) => this.mapRow(row));
}
markGotResponse(messageId: number): void {
this.markResponseStmt.run(Date.now(), messageId);
this.markResponseStmt.run(Date.now(), this.accountId, messageId);
}
markGotImage(messageId: number): void {
@@ -585,11 +613,11 @@ export class MessagesRepository {
}
markChecked(messageId: number): void {
this.markCheckedStmt.run(Date.now(), messageId);
this.markCheckedStmt.run(Date.now(), this.accountId, messageId);
}
hasSentTo(recipient: string): boolean {
return this.hasSentToStmt.get(recipient) !== undefined;
return this.hasSentToStmt.get(this.accountId, recipient) !== undefined;
}
getStrategyStats(): StrategyStats[] {
@@ -600,9 +628,9 @@ export class MessagesRepository {
SUM(CASE WHEN got_response = 1 THEN 1 ELSE 0 END) as total_responses,
SUM(CASE WHEN got_image = 1 THEN 1 ELSE 0 END) as total_images
FROM sent_messages
WHERE strategy_id IS NOT NULL
WHERE account_id = ? AND strategy_id IS NOT NULL
GROUP BY strategy_id
`).all() as Record<string, unknown>[];
`).all(this.accountId) as Record<string, unknown>[];
return rows.map(row => ({
strategyId: row.strategy_id as string,
@@ -626,10 +654,11 @@ export class MessagesRepository {
SUM(CASE WHEN sm.got_response = 1 THEN 1 ELSE 0 END) as total_responses,
SUM(CASE WHEN sm.got_image = 1 THEN 1 ELSE 0 END) as total_images
FROM sent_messages sm
JOIN processed_posts pp ON sm.post_id = pp.id
JOIN processed_posts pp ON sm.account_id = pp.account_id AND sm.post_id = pp.id
WHERE sm.account_id = ?
GROUP BY pp.subreddit
ORDER BY total_sent DESC
`).all() as Record<string, unknown>[];
`).all(this.accountId) as Record<string, unknown>[];
return rows.map(row => ({
subreddit: row.subreddit as string,
@@ -654,11 +683,11 @@ export class MessagesRepository {
SUM(CASE WHEN got_response = 1 THEN 1 ELSE 0 END) as total_responses,
SUM(CASE WHEN got_image = 1 THEN 1 ELSE 0 END) as total_images
FROM sent_messages
WHERE sent_at > ?
WHERE account_id = ? AND sent_at > ?
AND (post_id IS NOT NULL OR strategy_id IS NOT NULL)
GROUP BY day
ORDER BY day ASC
`).all(cutoff) as Record<string, unknown>[];
`).all(this.accountId, cutoff) as Record<string, unknown>[];
return rows.map(row => ({
date: row.day as string,
@@ -704,14 +733,18 @@ export class SuccessfulMessagesRepository {
private countStmt: Database.Statement;
private existsByMessageStmt: Database.Statement;
constructor(private db: Database.Database) {
constructor(
private db: Database.Database,
private readonly accountId: string = LEGACY_UNASSIGNED_ACCOUNT_ID,
) {
this.insertStmt = db.prepare(`
INSERT INTO successful_messages (sent_message_id, original_message, image_analysis, post_title, response_snippet, subreddit, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
INSERT INTO successful_messages (account_id, sent_message_id, original_message, image_analysis, post_title, response_snippet, subreddit, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
`);
this.getRecentStmt = db.prepare(`
SELECT * FROM successful_messages ORDER BY created_at DESC, id DESC LIMIT ?
SELECT * FROM successful_messages
WHERE account_id = ? ORDER BY created_at DESC, id DESC LIMIT ?
`);
this.getAttributedRecentStmt = db.prepare(`
@@ -730,14 +763,15 @@ export class SuccessfulMessagesRepository {
SELECT *,
CASE WHEN subreddit = ? THEN 1 ELSE 0 END as subreddit_match
FROM successful_messages
WHERE account_id = ?
ORDER BY subreddit_match DESC, created_at DESC
LIMIT ?
`);
this.countStmt = db.prepare(`SELECT COUNT(*) as count FROM successful_messages`);
this.countStmt = db.prepare(`SELECT COUNT(*) as count FROM successful_messages WHERE account_id = ?`);
this.existsByMessageStmt = db.prepare(`
SELECT 1 FROM successful_messages WHERE original_message = ? LIMIT 1
SELECT 1 FROM successful_messages WHERE account_id = ? AND original_message = ? LIMIT 1
`);
}
@@ -750,6 +784,7 @@ export class SuccessfulMessagesRepository {
subreddit: string | null;
}): void {
this.insertStmt.run(
this.accountId,
data.sentMessageId,
data.originalMessage,
data.imageAnalysis,
@@ -761,11 +796,11 @@ export class SuccessfulMessagesRepository {
}
existsByMessage(originalMessage: string): boolean {
return this.existsByMessageStmt.get(originalMessage) !== undefined;
return this.existsByMessageStmt.get(this.accountId, originalMessage) !== undefined;
}
getRecent(limit: number = 5): SuccessfulMessageRecord[] {
const rows = this.getRecentStmt.all(limit) as Record<string, unknown>[];
const rows = this.getRecentStmt.all(this.accountId, limit) as Record<string, unknown>[];
return rows.map((row) => this.mapRow(row));
}
@@ -776,14 +811,14 @@ export class SuccessfulMessagesRepository {
getWeightedExamples(subreddit: string | null, limit: number = 3): SuccessfulMessageRecord[] {
if (subreddit) {
const rows = this.getBySubredditStmt.all(subreddit, limit) as Record<string, unknown>[];
const rows = this.getBySubredditStmt.all(subreddit, this.accountId, limit) as Record<string, unknown>[];
return rows.map((row) => this.mapRow(row));
}
return this.getRecent(limit);
}
count(): number {
const result = this.countStmt.get() as { count: number };
const result = this.countStmt.get(this.accountId) as { count: number };
return result.count;
}
@@ -807,47 +842,55 @@ export class ConversationStateRepository {
private upsertStmt: Database.Statement;
private getStmt: Database.Statement;
constructor(private db: Database.Database) {
constructor(
private db: Database.Database,
private readonly accountId: string = LEGACY_UNASSIGNED_ACCOUNT_ID,
) {
this.upsertStmt = db.prepare(`
INSERT INTO conversation_states (conversation_id, state_json, updated_at)
VALUES (?, ?, ?)
ON CONFLICT(conversation_id) DO UPDATE SET
INSERT INTO conversation_states (account_id, conversation_id, state_json, updated_at)
VALUES (?, ?, ?, ?)
ON CONFLICT(account_id, conversation_id) DO UPDATE SET
state_json = excluded.state_json,
updated_at = excluded.updated_at
`);
this.getStmt = db.prepare(`
SELECT state_json FROM conversation_states WHERE conversation_id = ?
SELECT state_json FROM conversation_states WHERE account_id = ? AND conversation_id = ?
`);
}
save(conversationId: string, state: ConversationState): void {
this.upsertStmt.run(conversationId, serializeConversationState(state), Date.now());
this.upsertStmt.run(this.accountId, conversationId, serializeConversationState(state), Date.now());
}
get(conversationId: string): ConversationState | undefined {
const row = this.getStmt.get(conversationId) as { state_json: string } | undefined;
const row = this.getStmt.get(this.accountId, conversationId) as { state_json: string } | undefined;
return row ? parseConversationState(row.state_json) ?? undefined : undefined;
}
}
/** Outcome labels are recorded independently from a raw inbox response. */
export class ConversationOutcomesRepository {
constructor(private db: Database.Database) {}
constructor(
private db: Database.Database,
private readonly accountId: string = LEGACY_UNASSIGNED_ACCOUNT_ID,
) {}
record(sentMessageId: number, outcome: ConversationOutcome, responseSnippet: string | null): void {
this.db.prepare(`
INSERT INTO conversation_outcomes (sent_message_id, outcome, response_snippet, recorded_at)
VALUES (?, ?, ?, ?)
ON CONFLICT(sent_message_id) DO UPDATE SET
INSERT INTO conversation_outcomes (account_id, sent_message_id, outcome, response_snippet, recorded_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(account_id, sent_message_id) DO UPDATE SET
outcome = excluded.outcome,
response_snippet = excluded.response_snippet,
recorded_at = excluded.recorded_at
`).run(sentMessageId, outcome, responseSnippet, Date.now());
`).run(this.accountId, sentMessageId, outcome, responseSnippet, Date.now());
}
get(sentMessageId: number): { outcome: ConversationOutcome; responseSnippet: string | null } | undefined {
const row = this.db.prepare(`SELECT outcome, response_snippet FROM conversation_outcomes WHERE sent_message_id = ?`)
.get(sentMessageId) as { outcome: ConversationOutcome; response_snippet: string | null } | undefined;
const row = this.db.prepare(`
SELECT outcome, response_snippet FROM conversation_outcomes
WHERE account_id = ? AND sent_message_id = ?
`).get(this.accountId, sentMessageId) as { outcome: ConversationOutcome; response_snippet: string | null } | undefined;
return row ? { outcome: row.outcome, responseSnippet: row.response_snippet } : undefined;
}
}
src/main-loop.ts
+28
View File
@@ -74,6 +74,10 @@ const ERROR_BACKOFF_MS = 60_000;
* SIGINT/SIGTERM. Resolves once shutdown completes.
*/
export async function runMainLoop(app: AppContainer): Promise<void> {
if (app.accountRuntimeManager) {
await runConcurrentAccountRuntimes(app);
return;
}
const {
config,
logger,
@@ -407,3 +411,27 @@ export async function runMainLoop(app: AppContainer): Promise<void> {
}
}
}
async function runConcurrentAccountRuntimes(app: AppContainer): Promise<void> {
const { logger, accountRuntimeManager, browserClient, db, webServer } = app;
logger.info(`Running ${accountRuntimeManager.list().length} independent Reddit account runtime(s)`);
await new Promise<void>((resolve) => {
let shuttingDown = false;
const shutdown = async (): Promise<void> => {
if (shuttingDown) return;
shuttingDown = true;
logger.info('Stopping all Reddit account runtimes...');
await accountRuntimeManager.stopAll();
webServer?.stop();
await browserClient.close();
db.close();
process.off('SIGINT', onSignal);
process.off('SIGTERM', onSignal);
logger.info('Goodbye!');
resolve();
};
const onSignal = (): void => { void shutdown(); };
process.on('SIGINT', onSignal);
process.on('SIGTERM', onSignal);
});
}
src/services/chat/cache.ts
+70 -2
View File
@@ -2,6 +2,7 @@ import fs from 'fs/promises';
import path from 'path';
import type { Logger } from '../../utils/logger.js';
import type { ActiveChat, ChatCacheFile, InvitationCooldown } from './types.js';
import type { AccountContentRepository } from '../../database/account-repositories.js';
export interface SyncMarkers {
nextBatch: string | null;
@@ -24,8 +25,13 @@ export class ChatCache {
private lastDeltaSyncAt: number | null = null;
private cooldown: InvitationCooldown | null = null;
constructor(private logger: Logger, filePath?: string) {
if (!filePath && process.env.NODE_ENV === 'test') {
constructor(
private logger: Logger,
filePath?: string,
private readonly databaseStore?: AccountContentRepository,
private accountId: string | null = null,
) {
if (!filePath && !databaseStore && process.env.NODE_ENV === 'test') {
// Without this guard a test that constructs a ChatCache (or a
// ChatService that builds one) would silently write to the prod
// ./data/chat-cache.json — concurrent test runs have already torn
@@ -37,6 +43,18 @@ export class ChatCache {
this.filePath = filePath ?? path.join(process.cwd(), 'data', 'chat-cache.json');
}
async switchAccount(accountId: string): Promise<void> {
if (this.accountId === accountId) return;
this.accountId = accountId;
this.chats = [];
this.lastFetchTime = 0;
this.nextBatch = null;
this.lastFullSyncAt = null;
this.lastDeltaSyncAt = null;
this.cooldown = null;
await this.load();
}
get(): ActiveChat[] {
return this.chats;
}
@@ -80,6 +98,20 @@ export class ChatCache {
}
async load(): Promise<void> {
if (this.databaseStore && this.accountId) {
const rows = this.databaseStore.listConversations(this.accountId);
this.chats = rows.flatMap((row) => {
try { return [JSON.parse(row.payloadJson) as ActiveChat]; } catch { return []; }
});
const state = this.databaseStore.getSyncState(this.accountId);
this.nextBatch = state?.nextBatch ?? null;
this.lastFullSyncAt = state?.lastFullSyncAt ?? null;
this.lastDeltaSyncAt = state?.lastDeltaSyncAt ?? null;
try { this.cooldown = state?.cooldownJson ? JSON.parse(state.cooldownJson) as InvitationCooldown : null; } catch { this.cooldown = null; }
this.lastFetchTime = Date.now();
this.logger.info(`Loaded ${this.chats.length} account-owned chats from the database`);
return;
}
try {
const raw = await fs.readFile(this.filePath, 'utf-8');
const parsed = JSON.parse(raw) as ChatCacheFile;
@@ -115,6 +147,10 @@ export class ChatCache {
* sync with `chats: []` and null markers.
*/
async persistCooldown(): Promise<void> {
if (this.databaseStore && this.accountId) {
this.persistDatabaseState();
return;
}
let existing: ChatCacheFile | null = null;
try {
const raw = await fs.readFile(this.filePath, 'utf-8');
@@ -143,6 +179,27 @@ export class ChatCache {
}
async save(chats: ActiveChat[] = this.chats): Promise<void> {
if (this.databaseStore && this.accountId) {
const retained = new Set<string>();
for (const chat of chats) {
const conversationId = chat.roomId ?? chat.username;
retained.add(conversationId);
this.databaseStore.upsertConversation({
accountId: this.accountId,
conversationId,
username: chat.username,
roomName: chat.roomName ?? null,
payloadJson: JSON.stringify(chat),
});
}
for (const existing of this.databaseStore.listConversations(this.accountId)) {
if (!retained.has(existing.conversationId)) {
this.databaseStore.deleteConversation(this.accountId, existing.conversationId);
}
}
this.persistDatabaseState();
return;
}
const payload: ChatCacheFile = {
version: 1,
savedAt: Date.now(),
@@ -159,6 +216,17 @@ export class ChatCache {
}
}
private persistDatabaseState(): void {
if (!this.databaseStore || !this.accountId) return;
this.databaseStore.saveSyncState({
accountId: this.accountId,
nextBatch: this.nextBatch,
lastFullSyncAt: this.lastFullSyncAt,
lastDeltaSyncAt: this.lastDeltaSyncAt,
cooldownJson: this.cooldown ? JSON.stringify(this.cooldown) : null,
});
}
/**
* Write the cache file via a temp+rename so a crashed/interrupted/concurrent
* writer can never leave the on-disk file half-written. rename(2) is atomic
src/services/chat/service.ts
+9
View File
@@ -256,6 +256,15 @@ export class ChatService {
}
}
async switchAccount(accountId: string): Promise<void> {
await this.enqueue(async () => {
await this.cache.switchAccount(accountId);
this.t2ToUsername.clear();
this.unresolvableRooms.clear();
this.invitationCooldown = this.cache.getCooldown();
});
}
async saveCacheExternal(): Promise<void> {
await this.cache.save();
}
src/services/media.ts
+59 -7
View File
@@ -4,6 +4,7 @@ import crypto from 'crypto';
import { z } from 'zod';
import type { Logger } from '../utils/logger.js';
import type { ActiveChat, ChatMedia, ChatMessage } from './chat.js';
import type { AccountContentRepository } from '../database/account-repositories.js';
export interface MediaDownloadResult {
buffer: Buffer;
@@ -15,6 +16,7 @@ export interface MediaDownloader {
}
export interface MediaRecord {
accountId?: string;
id: string;
chatUsername: string | null;
chatRoomId: string | null;
@@ -59,13 +61,16 @@ export class MediaService {
private downloader: MediaDownloader,
private logger: Logger,
mediaRoot?: string,
private readonly databaseStore?: AccountContentRepository,
private readonly getAccountId: () => string | null = () => null,
) {
this.mediaRoot = mediaRoot ?? path.join(process.cwd(), 'data', 'media');
this.indexPath = path.join(this.mediaRoot, 'index.json');
}
async attachMedia(chat: ActiveChat, messages: ScrapedChatMessage[]): Promise<ChatMessage[]> {
await this.loadIndex();
const databaseBacked = Boolean(this.databaseStore && this.getAccountId());
if (!databaseBacked) await this.loadIndex();
const updated: ChatMessage[] = [];
for (const message of messages) {
@@ -79,11 +84,26 @@ export class MediaService {
});
}
await this.saveIndex();
if (!databaseBacked) await this.saveIndex();
return updated;
}
async getMediaItems(): Promise<MediaRecord[]> {
const accountId = this.getAccountId();
if (this.databaseStore && accountId) {
return this.databaseStore.listMedia(accountId).map((item) => ({
accountId: item.accountId,
id: item.mediaId,
chatUsername: item.chatUsername,
chatRoomId: item.chatRoomId,
sender: item.sender,
timestamp: item.timestamp,
originalUrl: item.originalUrl,
fileName: item.fileName,
relativePath: item.relativePath,
createdAt: item.createdAt,
}));
}
await this.loadIndex();
return (this.index?.items || []).slice().sort((a, b) => b.timestamp - a.timestamp);
}
@@ -133,11 +153,23 @@ export class MediaService {
// Try the full-resolution URL first, fall back to original
const fullUrl = this.resolveFullMediaUrl(url);
const id = crypto.createHash('sha1').update(fullUrl).digest('hex');
const accountId = this.getAccountId();
const id = crypto.createHash('sha1').update(`${accountId ?? 'legacy'}\0${fullUrl}`).digest('hex');
if (this.databaseStore && accountId) {
const existingRecord = this.databaseStore.getMedia(accountId, id);
if (existingRecord) {
try {
await fs.access(path.join(this.mediaRoot, existingRecord.relativePath));
return { ...existingRecord, id: existingRecord.mediaId };
} catch {
this.databaseStore.deleteMedia(accountId, id);
}
}
}
const existing = this.index?.items.find((item) => item.id === id);
if (existing) {
// Verify the file still exists on disk
const existingPath = path.join(this.mediaRoot, existing.fileName);
const existingPath = path.join(this.mediaRoot, existing.relativePath);
try {
await fs.access(existingPath);
return existing;
@@ -165,13 +197,17 @@ export class MediaService {
}
const fileName = `${id}${extension}`;
const relativePath = fileName;
const fullPath = path.join(this.mediaRoot, fileName);
const accountDirectory = accountId
? crypto.createHash('sha256').update(accountId).digest('hex').slice(0, 16)
: '';
const relativePath = accountDirectory ? path.join(accountDirectory, fileName) : fileName;
const fullPath = path.join(this.mediaRoot, relativePath);
await fs.mkdir(this.mediaRoot, { recursive: true });
await fs.mkdir(path.dirname(fullPath), { recursive: true });
await fs.writeFile(fullPath, download.buffer);
const record: MediaRecord = {
accountId: accountId ?? undefined,
id,
chatUsername: chat.username || null,
chatRoomId: chat.roomId || null,
@@ -183,6 +219,22 @@ export class MediaService {
createdAt: Date.now(),
};
if (this.databaseStore && accountId) {
this.databaseStore.upsertMedia({
accountId,
mediaId: id,
chatUsername: record.chatUsername,
chatRoomId: record.chatRoomId,
sender: record.sender,
timestamp: record.timestamp,
originalUrl: record.originalUrl,
fileName: record.fileName,
relativePath: record.relativePath,
createdAt: record.createdAt,
});
return record;
}
if (!this.index) {
this.index = { version: 1, savedAt: Date.now(), items: [record] };
} else {
src/services/reddit-account-runtime-builder.ts
+290
View File
@@ -0,0 +1,290 @@
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,
ConversationOutcomesRepository, RedditUsersRepository,
ActivityLogRepository,
} from '../database/repositories.js';
import { AccountContentRepository } from '../database/account-repositories.js';
import { RedditApiClient } from '../reddit/api.js';
import { OpenRouterClient } from '../ai/openrouter.js';
import { MonitorService } from './monitor.js';
import { ProcessorService, isSkip } from './processor.js';
import { ResponseCheckerService } from './response-checker.js';
import { StrategySelector } from './strategy-selector.js';
import { AccountAuthContext } from './account-auth-context.js';
import { ChatCache } from './chat/cache.js';
import {
ChatService, MatrixClient, MatrixMediaDownloader, MatrixTokenRefresher,
MatrixUsernameResolver, 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,
} from './reddit-account-runtime.js';
export interface ConcreteRuntimeFactoryOptions {
config: Config;
logger: Logger;
db: Database.Database;
isBotRunning?: () => boolean;
}
export function createConcreteRedditAccountRuntimeFactory(
options: ConcreteRuntimeFactoryOptions,
): RedditAccountRuntimeFactory {
return new RedditAccountRuntimeFactory((session) => buildRuntime(options, session));
}
async function buildRuntime(
{ config, logger, db, isBotRunning = () => true }: ConcreteRuntimeFactoryOptions,
session: VerifiedAccountSession,
): Promise<RedditAccountRuntime> {
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);
if (await matrixClient.whoami() !== session.matrixUserId) {
throw new Error(`Stored Matrix session no longer verifies for u/${session.redditUsername}`);
}
const accountAuth = new AccountAuthContext(logger);
accountAuth.markAuthenticated({
accountId: session.accountId,
redditUsername: identity.username,
matrixUserId: session.matrixUserId,
});
accountAuth.setAuthenticator(async () => {
const current = await browserClient.getAuthenticatedIdentity();
if (!current || matrixRedditId(session.matrixUserId) !== current.accountId.toLowerCase().replace(/^t2_/, '')) return null;
return await matrixClient.whoami() === session.matrixUserId
? { accountId: session.accountId, redditUsername: current.username, matrixUserId: session.matrixUserId }
: 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 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);
redditApi.setBrowserContext(browserClient.getContext()!);
const chatPage = await browserClient.getContext()!.newPage();
const aiClient = new OpenRouterClient(accountConfig, logger);
const monitorService = new MonitorService(redditApi, postsRepo, messagesRepo, accountConfig, logger);
const processorService = new ProcessorService(redditApi, aiClient, accountConfig, logger);
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,
new MatrixTokenRefresher(), usernameResolver, chatCache,
messagesRepo, redditUsersRepo, accountAuth,
);
await chatService.loadCache();
const mediaSyncService = new MediaSyncService(chatService, logger, accountConfig.media.syncIntervalMs, accountAuth);
const responseChecker = new ResponseCheckerService(
chatPage, messagesRepo, successfulMessagesRepo, postsRepo, logger,
accountAuth, conversationOutcomesRepo,
);
const rateLimiter = new RateLimiter(accountConfig, logger, undefined, {
sentMessageHistory: async (since) => messagesRepo.getSentTimestampsSince(since),
});
await rateLimiter.initialize();
const resources: RedditAccountRuntimeResources = {
browserProfileDir, browserClient, accountAuth, matrixClient, chatCache, redditApi,
postsRepo, messagesRepo, successfulMessagesRepo, deliveryOutboxRepo,
conversationStateRepo, conversationOutcomesRepo, redditUsersRepo, accountContentRepo,
activityLogRepo,
monitorService, processorService, responseChecker, chatService, mediaService,
mediaSyncService, rateLimiter,
};
const lifecycle = new AccountRuntimeLoop(resources, accountConfig, logger, isBotRunning);
return new RedditAccountRuntime(session, resources, lifecycle);
} catch (error) {
await browserClient.close();
throw error;
}
}
class AccountRuntimeLoop {
private controller: AbortController | null = null;
private loop: Promise<void> | null = null;
constructor(
private readonly resources: RedditAccountRuntimeResources,
private readonly config: Config,
private readonly logger: Logger,
private readonly isBotRunning: () => boolean,
) {}
start(): void {
if (this.loop) return;
this.controller = new AbortController();
this.resources.mediaSyncService.start();
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));
this.loop = this.run(this.controller.signal).finally(() => { this.loop = null; });
}
async stop(): Promise<void> {
this.controller?.abort();
this.resources.mediaSyncService.stop();
await this.loop?.catch((error) => {
if (!(error instanceof DOMException && error.name === 'AbortError')) throw error;
});
await this.resources.browserClient.close();
}
private async run(signal: AbortSignal): Promise<void> {
let nextResponseCheckAt = 0;
const responseInterval = Math.max(this.config.reddit.pollIntervalMs * 5, 60_000);
while (!signal.aborted) {
try {
if (!this.isBotRunning()) {
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() >= nextResponseCheckAt) {
nextResponseCheckAt = Date.now() + responseInterval;
await this.resources.responseChecker.checkForResponses();
}
if (!this.resources.chatService.getInvitationCooldown()) {
await this.processPosts(signal);
}
await delay(this.config.reddit.pollIntervalMs, signal);
} catch (error) {
if (signal.aborted) return;
this.logger.error(`Account runtime loop failed for ${this.resources.accountAuth.getSnapshot().redditUsername}:`, error);
await delay(60_000, signal);
}
}
}
private async processPosts(signal: AbortSignal): Promise<void> {
const { monitorService, postsRepo, messagesRepo, redditApi, processorService, rateLimiter, chatService, responseChecker } = this.resources;
const posts = await monitorService.getUnprocessedPosts();
for (const post of posts) {
if (signal.aborted) return;
const subreddit = this.config.reddit.subreddits.find((entry) => entry.name.toLowerCase() === post.subreddit.toLowerCase());
const isChat = subreddit?.type === 'chat';
const imageUrl = isChat ? null : redditApi.getImageUrl(post);
postsRepo.insert({
id: post.id, subreddit: post.subreddit, author: post.author, title: post.title,
postUrl: redditApi.getPostUrl(post), imageUrl, processedAt: Date.now(),
});
try {
if (
this.config.reddit.filterSellers
&& (this.config.reddit.sellerKeywords.length > 0 || this.config.reddit.sellerDomains.length > 0)
&& await redditApi.isSellerProfile(post.author, this.config.reddit.sellerKeywords, this.config.reddit.sellerDomains)
) {
postsRepo.setError(post.id, 'Seller profile detected');
this.resources.activityLogRepo.insert({ timestamp: Date.now(), type: 'skipped', message: `Skipped post by u/${post.author} (seller profile)`, recipient: post.author });
continue;
}
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 });
continue;
}
postsRepo.updateAnalysis(post.id, processed.imageAnalysis, processed.generatedMessage);
await rateLimiter.throttle(async () => {
const chat = await chatService.startConversation(post.author, processed.generatedMessage, post.author_fullname);
if (!chat) return false;
postsRepo.markMessageSent(post.id);
messagesRepo.insert({
postId: post.id, recipient: post.author, subject: post.author,
message: processed.generatedMessage, sentAt: Date.now(),
strategyId: processed.strategyId, chatRoomId: chat.roomId,
});
this.resources.activityLogRepo.insert({
timestamp: Date.now(), type: 'sent', message: `Sent message to u/${post.author} (r/${post.subreddit})`,
recipient: post.author, imageUrl: imageUrl ?? undefined, postTitle: post.title, postUrl: redditApi.getPostUrl(post),
});
void chatService.refreshTopChats(5);
return true;
}, (success) => success, signal);
} catch (error) {
if (signal.aborted) return;
postsRepo.setError(post.id, error instanceof Error ? error.message : String(error));
this.resources.activityLogRepo.insert({ timestamp: Date.now(), type: 'error', message: `Failed processing post by u/${post.author}`, recipient: post.author });
this.logger.error(`Account runtime failed processing post ${post.id}:`, error);
}
}
}
}
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 });
});
}
src/services/reddit-account-runtime.ts
+378
View File
@@ -0,0 +1,378 @@
import path from 'node:path';
import type { Cookie } from 'playwright';
import { z } from 'zod';
import type { BrowserClient } from '../browser/client.js';
import type {
RedditAccountPrivateRecord,
RedditAccountsRepository,
AccountContentRepository,
} from '../database/account-repositories.js';
import type {
ConversationOutcomesRepository,
ActivityLogRepository,
ConversationStateRepository,
DeliveryOutboxRepository,
MessagesRepository,
PostsRepository,
RedditUsersRepository,
SuccessfulMessagesRepository,
} from '../database/repositories.js';
import type { RedditApiClient } from '../reddit/api.js';
import type { RateLimiter } from '../utils/rate-limiter.js';
import type { AccountAuthContext } from './account-auth-context.js';
import type { ChatCache } from './chat/cache.js';
import type { MatrixClient } from './chat/matrix/client.js';
import type { MatrixCredentials } from './chat/matrix/types.js';
import type { ChatService } from './chat/service.js';
import type { MediaService } from './media.js';
import type { MediaSyncService } from './media-sync.js';
import type { MonitorService } from './monitor.js';
import type { ProcessorService } from './processor.js';
import type { ResponseCheckerService } from './response-checker.js';
/** Credentials passed to a factory. They must never be exposed by list/snapshot APIs. */
export interface VerifiedAccountSession {
accountId: string;
redditUsername: string;
matrixUserId: string;
cookies: Cookie[];
matrixCredentials: MatrixCredentials;
updatedAt: number;
}
/**
* The complete set of stateful dependencies owned by one account runtime.
*
* Nothing in this object may be shared with another account runtime except the
* underlying SQLite connection and stateless configuration/AI clients. In
* particular, BrowserClient, AccountAuthContext, ChatCache, MatrixClient,
* RedditApiClient and RateLimiter are intentionally per-account instances.
*/
export interface RedditAccountRuntimeResources {
browserProfileDir: string;
browserClient: BrowserClient;
accountAuth: AccountAuthContext;
matrixClient: MatrixClient;
chatCache: ChatCache;
redditApi: RedditApiClient;
postsRepo: PostsRepository;
messagesRepo: MessagesRepository;
successfulMessagesRepo: SuccessfulMessagesRepository;
deliveryOutboxRepo: DeliveryOutboxRepository;
conversationStateRepo: ConversationStateRepository;
conversationOutcomesRepo: ConversationOutcomesRepository;
redditUsersRepo: RedditUsersRepository;
accountContentRepo: AccountContentRepository;
activityLogRepo: ActivityLogRepository;
monitorService: MonitorService;
processorService: ProcessorService;
responseChecker: ResponseCheckerService;
chatService: ChatService;
mediaService: MediaService;
mediaSyncService: MediaSyncService;
rateLimiter: RateLimiter;
}
export interface RedditAccountRuntimeLifecycle {
/** Start this account's main loop, response schedule and media sync. */
start(): void | Promise<void>;
/** Stop and await this account's schedules/loop, then close its browser. */
stop(): void | Promise<void>;
}
export type RedditAccountRuntimeState =
| 'created'
| 'starting'
| 'running'
| 'stopping'
| 'stopped'
| 'failed';
export interface RedditAccountRuntimeSnapshot {
accountId: string;
redditUsername: string;
matrixUserId: string;
state: RedditAccountRuntimeState;
updatedAt: number;
failure: string | null;
}
/**
* One immutable Reddit identity and every mutable service associated with it.
* There is deliberately no switchAccount method: a session refresh replaces
* the matching runtime instead of mutating shared live state.
*/
export class RedditAccountRuntime {
private state: RedditAccountRuntimeState = 'created';
private failure: string | null = null;
private transition: Promise<void> = Promise.resolve();
constructor(
readonly session: VerifiedAccountSession,
readonly resources: RedditAccountRuntimeResources,
private readonly lifecycle: RedditAccountRuntimeLifecycle,
) {
if (session.accountId !== session.matrixUserId) {
throw new Error('Runtime account ID must match the verified Matrix user ID');
}
if (!session.accountId || !session.redditUsername) {
throw new Error('Runtime requires a verified Reddit and Matrix identity');
}
}
start(): Promise<void> {
return this.enqueue(async () => {
if (this.state === 'running') return;
this.state = 'starting';
this.failure = null;
try {
await this.lifecycle.start();
this.state = 'running';
} catch (error) {
this.state = 'failed';
this.failure = errorMessage(error);
try {
await this.lifecycle.stop();
} catch (cleanupError) {
throw new AggregateError([error, cleanupError], 'Account runtime failed to start and clean up');
}
throw error;
}
});
}
stop(): Promise<void> {
return this.enqueue(async () => {
if (this.state === 'created' || this.state === 'stopped') {
this.state = 'stopped';
return;
}
this.state = 'stopping';
try {
await this.lifecycle.stop();
this.state = 'stopped';
} catch (error) {
this.state = 'failed';
this.failure = errorMessage(error);
throw error;
}
});
}
snapshot(): RedditAccountRuntimeSnapshot {
return {
accountId: this.session.accountId,
redditUsername: this.session.redditUsername,
matrixUserId: this.session.matrixUserId,
state: this.state,
updatedAt: this.session.updatedAt,
failure: this.failure,
};
}
private enqueue(operation: () => Promise<void>): Promise<void> {
const result = this.transition.then(operation, operation);
this.transition = result.catch(() => undefined);
return result;
}
}
export type RedditAccountRuntimeBuilder = (
session: VerifiedAccountSession,
) => Promise<RedditAccountRuntime>;
/** Narrow factory wrapper used as the bootstrap integration seam. */
export class RedditAccountRuntimeFactory {
constructor(private readonly buildRuntime: RedditAccountRuntimeBuilder) {}
async create(record: RedditAccountPrivateRecord): Promise<RedditAccountRuntime> {
return this.buildRuntime(parseStoredSession(record));
}
async createSession(session: VerifiedAccountSession): Promise<RedditAccountRuntime> {
return this.buildRuntime(session);
}
}
export interface RedditAccountRuntimeManagerSnapshot {
runtimes: RedditAccountRuntimeSnapshot[];
}
/**
* Owns all enabled account runtimes. Work for different account IDs runs in
* parallel; operations for the same ID are serialized to avoid duplicate
* loops and browser-profile contention.
*/
export class RedditAccountRuntimeManager {
private readonly runtimes = new Map<string, RedditAccountRuntime>();
private readonly accountOperations = new Map<string, Promise<void>>();
constructor(
private readonly accounts: Pick<RedditAccountsRepository, 'listPublic' | 'get'>,
private readonly factory: RedditAccountRuntimeFactory,
) {}
async startEnabled(): Promise<RedditAccountRuntimeManagerSnapshot> {
const enabled = this.accounts.listPublic().filter(isEnabledAccount);
await Promise.all(enabled.map((account) => this.start(account.accountId)));
return this.snapshot();
}
start(accountId: string): Promise<void> {
return this.forAccount(accountId, async () => {
const existing = this.runtimes.get(accountId);
if (existing?.snapshot().state === 'running') return;
const account = this.requireEnabledAccount(accountId);
const runtime = existing ?? await this.factory.create(account);
await runtime.start();
this.runtimes.set(accountId, runtime);
});
}
/**
* Rebuild only the refreshed account. Other account loops are untouched.
* The old runtime is stopped before concrete construction because a builder
* may open the account's persistent browser profile immediately.
*/
replace(accountId: string): Promise<void> {
return this.forAccount(accountId, async () => {
const account = this.requireEnabledAccount(accountId);
const previous = this.runtimes.get(accountId);
if (previous) await previous.stop();
try {
// Concrete construction opens the account's persistent browser
// profile, so the prior owner must be closed before factory.create.
const replacement = await this.factory.create(account);
await replacement.start();
this.runtimes.set(accountId, replacement);
} catch (error) {
if (previous) {
try {
const restored = await this.factory.createSession(previous.session);
await restored.start();
this.runtimes.set(accountId, restored);
} catch (rollbackError) {
this.runtimes.delete(accountId);
throw new AggregateError(
[error, rollbackError],
`Could not replace or restore Reddit account runtime ${accountId}`,
);
}
} else {
this.runtimes.delete(accountId);
}
throw error;
}
});
}
stop(accountId: string): Promise<void> {
return this.forAccount(accountId, async () => {
const runtime = this.runtimes.get(accountId);
if (!runtime) return;
await runtime.stop();
this.runtimes.delete(accountId);
});
}
async stopAll(): Promise<void> {
await Promise.all([...this.runtimes.keys()].map((accountId) => this.stop(accountId)));
}
get(accountId: string): RedditAccountRuntime | null {
return this.runtimes.get(accountId) ?? null;
}
list(): RedditAccountRuntimeSnapshot[] {
return this.snapshot().runtimes;
}
snapshot(): RedditAccountRuntimeManagerSnapshot {
return {
runtimes: [...this.runtimes.values()]
.map((runtime) => runtime.snapshot())
.sort((left, right) => left.redditUsername.localeCompare(right.redditUsername)),
};
}
private requireEnabledAccount(accountId: string): RedditAccountPrivateRecord {
const account = this.accounts.get(accountId);
if (!account || !isEnabledAccount(account)) {
throw new Error(`Reddit account ${accountId} is not enabled with a verified session`);
}
return account;
}
private forAccount(accountId: string, operation: () => Promise<void>): Promise<void> {
const previous = this.accountOperations.get(accountId) ?? Promise.resolve();
const result = previous.then(operation, operation);
const settled = result.catch(() => undefined).finally(() => {
if (this.accountOperations.get(accountId) === settled) {
this.accountOperations.delete(accountId);
}
});
this.accountOperations.set(accountId, settled);
return result;
}
}
/**
* Deterministic, filesystem-safe profile path for one immutable account ID.
* Bootstrap should clone config and set browser.sessionDir to this value
* before constructing that account's BrowserClient.
*/
export function accountBrowserProfileDir(baseSessionDir: string, accountId: string): string {
if (!accountId) throw new Error('Browser profile requires an account ID');
const encodedAccountId = Buffer.from(accountId, 'utf8').toString('base64url');
return path.join(baseSessionDir, 'accounts', encodedAccountId);
}
function parseStoredSession(record: RedditAccountPrivateRecord): VerifiedAccountSession {
const cookies = StoredCookiesSchema.parse(JSON.parse(record.cookiesJson)) as Cookie[];
const matrixCredentials = StoredMatrixCredentialsSchema.parse(
JSON.parse(record.matrixCredentialsJson),
) as MatrixCredentials;
if (record.matrixUserId !== record.accountId || matrixCredentials.userId !== record.accountId) {
throw new Error('Stored Reddit account and Matrix identities do not match');
}
return {
accountId: record.accountId,
redditUsername: record.redditUsername,
matrixUserId: record.accountId,
cookies,
matrixCredentials,
updatedAt: record.updatedAt,
};
}
const StoredCookiesSchema = z.array(z.object({
name: z.string().min(1),
value: z.string(),
domain: z.string().optional(),
path: z.string().optional(),
expires: z.number().optional(),
httpOnly: z.boolean().optional(),
secure: z.boolean().optional(),
sameSite: z.enum(['Strict', 'Lax', 'None']).optional(),
url: z.string().url().optional(),
}));
const StoredMatrixCredentialsSchema = z.object({
homeserver: z.string().url(),
accessToken: z.string().min(1),
userId: z.string().min(1),
capturedAt: z.number(),
});
function isEnabledAccount(account: {
status: string;
isActive: boolean;
}): boolean {
return account.status === 'verified' && account.isActive;
}
function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : String(error);
}
src/services/reddit-accounts.ts
+96
View File
@@ -0,0 +1,96 @@
import { z } from 'zod';
import type { Cookie } from 'playwright';
import type { RedditAccountsRepository, RedditAccountPublicRecord } from '../database/account-repositories.js';
import type { AccountAuthContext } from './account-auth-context.js';
import type { RedditSessionImporter } from './reddit-session-importer.js';
import type { MatrixCredentials } from './chat/matrix/types.js';
import type { RedditAccountRuntimeManager } from './reddit-account-runtime.js';
import type { BrowserClient } from '../browser/client.js';
const StoredCookiesSchema = z.array(z.object({
name: z.string(), value: z.string(), domain: z.string(), path: z.string(),
expires: z.number(), httpOnly: z.boolean(), secure: z.boolean(),
sameSite: z.enum(['Strict', 'Lax', 'None']),
}));
const StoredMatrixCredentialsSchema = z.object({
homeserver: z.string().url(),
accessToken: z.string().min(1),
userId: z.string().min(1),
capturedAt: z.number(),
});
export interface RedditAccountSummary {
accountId: string;
redditUsername: string;
matrixUserId: string;
isRunning: boolean;
status: 'authenticated' | 'login-required';
createdAt: number;
updatedAt: number;
lastImportedAt: number;
}
export class RedditAccountsService {
constructor(
private readonly accounts: RedditAccountsRepository,
private readonly importer: RedditSessionImporter,
private readonly auth: AccountAuthContext,
private readonly runtimes?: RedditAccountRuntimeManager,
private readonly verificationBrowser?: BrowserClient,
) {}
list(): RedditAccountSummary[] {
return this.accounts.listPublic()
.filter((account) => account.status !== 'legacy-unassigned')
.map((account) => this.toSummary(account));
}
get(accountId: string): RedditAccountSummary | null {
return this.list().find((account) => account.accountId === accountId) ?? null;
}
async activate(accountId: string): Promise<RedditAccountSummary> {
const stored = this.accounts.get(accountId);
if (!stored || stored.status === 'legacy-unassigned') {
throw new Error('Reddit account not found');
}
const cookies = StoredCookiesSchema.parse(JSON.parse(stored.cookiesJson)) as Cookie[];
const credentials = StoredMatrixCredentialsSchema.parse(
JSON.parse(stored.matrixCredentialsJson),
) as MatrixCredentials;
return this.importSession(cookies, credentials);
}
async importSession(cookies: Cookie[], credentials: MatrixCredentials): Promise<RedditAccountSummary> {
try {
const verified = await this.importer.import(cookies, credentials);
if (this.runtimes) {
const existing = this.runtimes.get(verified.accountId);
if (existing) await this.runtimes.replace(verified.accountId);
else await this.runtimes.start(verified.accountId);
}
const account = this.get(verified.accountId);
if (!account) throw new Error('Activated account was not persisted');
return account;
} finally {
await this.verificationBrowser?.close();
}
}
private toSummary(account: RedditAccountPublicRecord): RedditAccountSummary {
const runtime = this.runtimes?.get(account.accountId);
const running = runtime?.snapshot().state === 'running';
return {
accountId: account.accountId,
redditUsername: account.redditUsername,
matrixUserId: account.matrixUserId ?? account.accountId,
isRunning: running ?? (account.accountId === this.auth.getSnapshot().accountId),
status: account.status === 'verified' && (running ?? account.accountId === this.auth.getSnapshot().accountId)
? 'authenticated'
: 'login-required',
createdAt: account.createdAt,
updatedAt: account.updatedAt,
lastImportedAt: account.lastImportedAt ?? account.updatedAt,
};
}
}
src/services/reddit-session-importer.ts
+39 -4
View File
@@ -7,15 +7,32 @@ interface BrowserSessionTarget {
importRedditCookies(cookies: Cookie[]): Promise<void>;
isSessionValid(): Promise<boolean>;
getAuthenticatedUsername(): Promise<string | null>;
getAuthenticatedIdentity?(): Promise<{ username: string; accountId: string } | null>;
}
interface ChatSessionTarget {
setMatrixClient(client: MatrixClient): void;
switchAccount?(accountId: string): Promise<void>;
}
type MatrixClientFactory = (credentials: MatrixCredentials) => MatrixClient;
type CredentialStore = (credentials: MatrixCredentials) => void;
export interface VerifiedRedditSessionStore {
save(input: {
accountId: string;
redditUsername: string;
matrixUserId: string;
cookies: Cookie[];
credentials: MatrixCredentials;
}): void;
}
function matrixRedditAccountId(matrixUserId: string): string | null {
const match = matrixUserId.match(/^@t2_([^:]+):reddit\.com$/i);
return match?.[1]?.toLowerCase() ?? null;
}
export class RedditSessionImporter {
constructor(
private readonly browser: BrowserSessionTarget,
@@ -23,9 +40,14 @@ export class RedditSessionImporter {
private readonly createMatrixClient: MatrixClientFactory,
private readonly saveCredentials: CredentialStore,
private readonly accountAuth: AccountAuthContext,
private readonly sessionStore?: VerifiedRedditSessionStore,
) {}
async import(cookies: Cookie[], credentials: MatrixCredentials): Promise<void> {
async import(cookies: Cookie[], credentials: MatrixCredentials): Promise<{
accountId: string;
redditUsername: string;
matrixUserId: string;
}> {
const homeserver = new URL(credentials.homeserver);
if (
homeserver.protocol !== 'https:' ||
@@ -44,15 +66,28 @@ export class RedditSessionImporter {
if (!(await this.browser.isSessionValid())) {
throw new Error('Imported cookies did not authenticate with Reddit');
}
const redditUsername = await this.browser.getAuthenticatedUsername();
const identity = this.browser.getAuthenticatedIdentity
? await this.browser.getAuthenticatedIdentity()
: null;
const redditUsername = identity?.username ?? await this.browser.getAuthenticatedUsername();
if (!redditUsername) throw new Error('Reddit did not return an authenticated account identity');
if (identity) {
const matrixAccountId = matrixRedditAccountId(userId);
if (!matrixAccountId || matrixAccountId !== identity.accountId.toLowerCase().replace(/^t2_/, '')) {
throw new Error('Reddit cookies and Matrix credentials belong to different accounts');
}
}
this.saveCredentials(credentials);
this.chat.setMatrixClient(matrixClient);
this.accountAuth.markAuthenticated({
const account = {
accountId: userId,
redditUsername,
matrixUserId: userId,
});
};
this.sessionStore?.save({ ...account, cookies, credentials });
await this.chat.switchAccount?.(account.accountId);
this.accountAuth.markAuthenticated(account);
return account;
}
}
src/web/context.ts
+4
View File
@@ -20,6 +20,8 @@ import type { RuntimeConfig } from './types.js';
import type { RedditSessionImporter } from '../services/reddit-session-importer.js';
import type { AccountAuthContext } from '../services/account-auth-context.js';
import type { ConfigStore } from '../config/persistence.js';
import type { RedditAccountsService } from '../services/reddit-accounts.js';
import type { RedditAccountRuntimeManager } from '../services/reddit-account-runtime.js';
/** Bag of everything the route modules might need. */
export interface WebContext {
@@ -48,6 +50,8 @@ export interface WebContext {
accountAuth?: AccountAuthContext;
conversationState?: ConversationStateRepository;
configStore?: ConfigStore;
redditAccounts?: RedditAccountsService;
accountRuntimes?: RedditAccountRuntimeManager;
};
/** Bot pause/resume toggle read by the main loop. */
src/web/routes/chats.ts
+66 -1
View File
@@ -12,7 +12,15 @@ export function createChatsRouter(ctx: WebContext): Router {
return;
}
try {
const all = await ctx.services.chat.getCachedConversations();
const runtimes = runningRuntimes(ctx);
const all = runtimes.length > 0
? (await Promise.all(runtimes.map(async (runtime) =>
(await runtime.resources.chatService.getCachedConversations()).map((chat) => ({
...chat,
owner: { accountId: runtime.session.accountId, redditUsername: runtime.session.redditUsername },
})),
))).flat()
: await ctx.services.chat.getCachedConversations();
const cutoff = Date.now() - ACTIVE_WINDOW_MS;
res.json(all.filter((c) => c.lastMessageTime >= cutoff));
} catch (error) {
@@ -23,6 +31,21 @@ export function createChatsRouter(ctx: WebContext): Router {
/** Status snapshot for the dashboard's sync controls. */
router.get('/chats/sync', (_req, res) => {
const runtimes = runningRuntimes(ctx);
if (runtimes.length > 0) {
const statuses = runtimes.map((runtime) => runtime.resources.chatService.getSyncStatus());
const active = statuses.find((status) => status.syncing);
res.json({
syncing: Boolean(active),
syncingKind: active?.syncingKind ?? null,
lastFullSyncAt: newest(statuses.map((status) => status.lastFullSyncAt)),
lastDeltaSyncAt: newest(statuses.map((status) => status.lastDeltaSyncAt)),
hasEverFullSynced: statuses.every((status) => status.hasEverFullSynced),
cachedChats: statuses.reduce((total, status) => total + status.cachedChats, 0),
progress: active?.progress ?? null,
});
return;
}
if (!ctx.services.chat) {
res.status(503).json({ error: 'Chat service not initialized' });
return;
@@ -35,6 +58,17 @@ export function createChatsRouter(ctx: WebContext): Router {
* Use this for routine refreshes.
*/
router.post('/chats/sync/delta', async (_req, res) => {
const runtimes = runningRuntimes(ctx);
if (runtimes.length > 0) {
const available = runtimes.filter((runtime) => !runtime.resources.chatService.getSyncStatus().syncing);
for (const runtime of available) {
void runtime.resources.chatService.refreshTopChats().catch((error) =>
ctx.logger.error(`Delta sync failed for u/${runtime.session.redditUsername}:`, error),
);
}
res.json(available.length > 0 ? { success: true, kind: 'delta' } : { success: false, reason: 'All account syncs are already running' });
return;
}
if (!ctx.services.chat) {
res.status(500).json({ error: 'Chat service not initialized' });
return;
@@ -64,6 +98,17 @@ export function createChatsRouter(ctx: WebContext): Router {
* appear in /sync.
*/
router.post('/chats/sync/full', async (_req, res) => {
const runtimes = runningRuntimes(ctx);
if (runtimes.length > 0) {
const available = runtimes.filter((runtime) => !runtime.resources.chatService.getSyncStatus().syncing);
for (const runtime of available) {
void runtime.resources.chatService.fetchAllMessages().catch((error) =>
ctx.logger.error(`Full sync failed for u/${runtime.session.redditUsername}:`, error),
);
}
res.json(available.length > 0 ? { success: true, kind: 'full' } : { success: false, reason: 'All account syncs are already running' });
return;
}
if (!ctx.services.chat) {
res.status(500).json({ error: 'Chat service not initialized' });
return;
@@ -87,6 +132,12 @@ export function createChatsRouter(ctx: WebContext): Router {
/** Back-compat alias for the old "fetch-all" endpoint. */
router.post('/chats/fetch-all', async (_req, res) => {
const runtimes = runningRuntimes(ctx);
if (runtimes.length > 0) {
const chats = (await Promise.all(runtimes.map((runtime) => runtime.resources.chatService.fetchAllMessages()))).flat();
res.json({ success: true, count: chats.length });
return;
}
if (!ctx.services.chat) {
res.status(500).json({ error: 'Chat service not initialized' });
return;
@@ -102,3 +153,17 @@ export function createChatsRouter(ctx: WebContext): Router {
return router;
}
function runningRuntimes(ctx: WebContext) {
return (ctx.services.accountRuntimes?.list() ?? [])
.filter((snapshot) => snapshot.state === 'running')
.flatMap((snapshot) => {
const runtime = ctx.services.accountRuntimes?.get(snapshot.accountId);
return runtime ? [runtime] : [];
});
}
function newest(values: Array<number | null>): number | null {
const present = values.filter((value): value is number => value !== null);
return present.length > 0 ? Math.max(...present) : null;
}
src/web/routes/conversations.ts
+128 -45
View File
@@ -7,26 +7,60 @@ import {
mergeConversationState,
} from '../../services/conversation-state.js';
type OwnedActiveChat = ActiveChat & { owner?: { accountId: string; redditUsername: string } };
export function createConversationsRouter(ctx: WebContext): Router {
const router = Router();
async function findConversation(id: string): Promise<ActiveChat | undefined> {
if (!ctx.services.chat) return undefined;
const conversations = await ctx.services.chat.getCachedConversations();
return conversations.find(
(c) => c.roomId === id || c.username === id || c.roomName === id,
);
function currentOwner() {
const snapshot = ctx.services.accountAuth?.getSnapshot();
return snapshot?.accountId && snapshot.redditUsername
? { accountId: snapshot.accountId, redditUsername: snapshot.redditUsername }
: undefined;
}
function getConversationIntelligence(conversation: ActiveChat, requestedIntent?: 'media-ask') {
function parseOwnedId(id: string): { accountId: string; conversationId: string } | null {
const separator = id.indexOf('~');
if (separator < 1) return null;
try {
return {
accountId: decodeURIComponent(id.slice(0, separator)),
conversationId: decodeURIComponent(id.slice(separator + 1)),
};
} catch { return null; }
}
async function findConversation(id: string): Promise<{
conversation: ActiveChat;
runtime: NonNullable<ReturnType<NonNullable<WebContext['services']['accountRuntimes']>['get']>> | null;
} | undefined> {
const owned = parseOwnedId(id);
if (!owned && (ctx.services.accountRuntimes?.list().length ?? 0) > 0) return undefined;
const runtime = owned ? ctx.services.accountRuntimes?.get(owned.accountId) ?? null : null;
if (owned && (!runtime || runtime.snapshot().state !== 'running')) return undefined;
const chat = runtime?.resources.chatService ?? ctx.services.chat;
if (!chat) return undefined;
const conversations = await chat.getCachedConversations();
const conversationId = owned?.conversationId ?? id;
const conversation = conversations.find(
(c) => c.roomId === conversationId || c.username === conversationId || c.roomName === conversationId,
);
return conversation ? { conversation, runtime } : undefined;
}
function getConversationIntelligence(
conversation: ActiveChat,
stateRepo = ctx.services.conversationState,
requestedIntent?: 'media-ask',
) {
const messages = (conversation.messages ?? []).map((message) => ({
sender: message.sender,
text: message.text,
isSelf: message.isSelf,
}));
const saved = conversation.roomId ? ctx.services.conversationState?.get(conversation.roomId) : undefined;
const saved = conversation.roomId ? stateRepo?.get(conversation.roomId) : undefined;
const state = mergeConversationState(saved, buildConversationState(messages));
if (conversation.roomId) ctx.services.conversationState?.save(conversation.roomId, state);
if (conversation.roomId) stateRepo?.save(conversation.roomId, state);
return { messages, state, intent: chooseConversationIntent(messages, state, requestedIntent) };
}
@@ -36,9 +70,18 @@ export function createConversationsRouter(ctx: WebContext): Router {
return;
}
try {
const conversations = await ctx.services.chat.getCachedConversations();
const runtimes = ctx.services.accountRuntimes?.list().filter((runtime) => runtime.state === 'running') ?? [];
const conversations: OwnedActiveChat[] = runtimes.length > 0
? (await Promise.all(runtimes.map(async (snapshot) => {
const runtime = ctx.services.accountRuntimes!.get(snapshot.accountId)!;
return (await runtime.resources.chatService.getCachedConversations()).map((conversation) => ({
...conversation,
owner: { accountId: snapshot.accountId, redditUsername: snapshot.redditUsername },
}));
}))).flat()
: await ctx.services.chat.getCachedConversations();
if (req.query.includeImageMetadata !== 'true') {
res.json(conversations);
res.json(conversations.map((conversation) => ({ ...conversation, owner: conversation.owner ?? currentOwner() })));
return;
}
const receivedMediaRooms = new Set<string>();
@@ -46,11 +89,18 @@ export function createConversationsRouter(ctx: WebContext): Router {
if (ctx.services.media) {
try {
const mediaItems = await ctx.services.media.getMediaItems();
for (const item of mediaItems) {
if (item.sender.toLowerCase() === 'me') continue;
if (item.chatRoomId) receivedMediaRooms.add(item.chatRoomId);
if (item.chatUsername) receivedMediaUsers.add(item.chatUsername.toLowerCase());
const mediaByAccount = runtimes.length > 0
? await Promise.all(runtimes.map(async (snapshot) => ({
accountId: snapshot.accountId,
items: await ctx.services.accountRuntimes!.get(snapshot.accountId)!.resources.mediaService.getMediaItems(),
})))
: [{ accountId: currentOwner()?.accountId ?? 'legacy', items: await ctx.services.media.getMediaItems() }];
for (const source of mediaByAccount) {
for (const item of source.items) {
if (item.sender.toLowerCase() === 'me') continue;
if (item.chatRoomId) receivedMediaRooms.add(`${source.accountId}:${item.chatRoomId}`);
if (item.chatUsername) receivedMediaUsers.add(`${source.accountId}:${item.chatUsername.toLowerCase()}`);
}
}
} catch (error) {
ctx.logger.warn('Failed to read media index for conversation markers:', error);
@@ -59,9 +109,10 @@ export function createConversationsRouter(ctx: WebContext): Router {
res.json(conversations.map((conversation) => ({
...conversation,
owner: conversation.owner ?? currentOwner(),
hasReceivedImages:
receivedMediaRooms.has(conversation.roomId ?? '')
|| receivedMediaUsers.has(conversation.username.toLowerCase())
receivedMediaRooms.has(`${conversation.owner?.accountId ?? currentOwner()?.accountId ?? 'legacy'}:${conversation.roomId ?? ''}`)
|| receivedMediaUsers.has(`${conversation.owner?.accountId ?? currentOwner()?.accountId ?? 'legacy'}:${conversation.username.toLowerCase()}`)
|| (conversation.messages ?? []).some(
(message) => !message.isSelf && (message.media?.length ?? 0) > 0,
),
@@ -80,13 +131,15 @@ export function createConversationsRouter(ctx: WebContext): Router {
try {
// Conversation reads never initiate a sync. Sync lifecycle events can
// invalidate this query, so starting a sync here would create a loop.
const syncStatus = ctx.services.chat.getSyncStatus();
const syncInFlight = syncStatus.syncing;
const conversation = await findConversation(decodeURIComponent(req.params.id));
if (!conversation) {
const resolved = await findConversation(decodeURIComponent(req.params.id));
if (!resolved) {
res.status(404).json({ error: 'Conversation not found' });
return;
}
const chatService = resolved.runtime?.resources.chatService ?? ctx.services.chat;
const syncStatus = chatService.getSyncStatus();
const syncInFlight = syncStatus.syncing;
const conversation = resolved.conversation;
// Delta sync only populates room summaries — full message history is
// fetched on demand the first time a conversation is opened, and
// re-fetched when the summary's last-activity timestamp is newer than
@@ -99,8 +152,8 @@ export function createConversationsRouter(ctx: WebContext): Router {
!conversation.messages || conversation.messages.length === 0 || hasNewerActivity;
if (needsLazyLoad && !syncInFlight) {
try {
await ctx.services.chat.loadConversationDetails(conversation);
await ctx.services.chat.saveCacheExternal();
await chatService.loadConversationDetails(conversation);
await chatService.saveCacheExternal();
} catch (err) {
ctx.logger.warn(`Lazy-load failed for conversation ${conversation.roomId}:`, err);
}
@@ -116,10 +169,11 @@ export function createConversationsRouter(ctx: WebContext): Router {
sentAt: number;
} | null = null;
try {
const messagesRepo = resolved.runtime?.resources.messagesRepo ?? ctx.repos.messages;
const sent = conversation.roomId
? ctx.repos.messages.getByChatRoomId(conversation.roomId)
?? ctx.repos.messages.getLatestByRecipient(conversation.username)
: ctx.repos.messages.getLatestByRecipient(conversation.username);
? messagesRepo.getByChatRoomId(conversation.roomId)
?? messagesRepo.getLatestByRecipient(conversation.username)
: messagesRepo.getLatestByRecipient(conversation.username);
if (sent && sent.postId) {
const url = sent.postUrl ?? (sent.subreddit
? `https://www.reddit.com/r/${sent.subreddit}/comments/${sent.postId}/`
@@ -139,6 +193,9 @@ export function createConversationsRouter(ctx: WebContext): Router {
res.json({
...conversation,
owner: resolved.runtime
? { accountId: resolved.runtime.session.accountId, redditUsername: resolved.runtime.session.redditUsername }
: currentOwner(),
triggeringPost,
syncInProgress: syncInFlight
? {
@@ -164,11 +221,12 @@ export function createConversationsRouter(ctx: WebContext): Router {
return;
}
try {
const conversation = await findConversation(decodeURIComponent(req.params.id));
if (!conversation) {
const resolved = await findConversation(decodeURIComponent(req.params.id));
if (!resolved) {
res.status(404).json({ error: 'Conversation not found' });
return;
}
const conversation = resolved.conversation;
if (!conversation.messages || conversation.messages.length === 0) {
res.status(400).json({ error: 'No messages in conversation to generate a reply from' });
return;
@@ -178,7 +236,7 @@ export function createConversationsRouter(ctx: WebContext): Router {
return;
}
const intelligence = getConversationIntelligence(conversation);
const intelligence = getConversationIntelligence(conversation, resolved.runtime?.resources.conversationStateRepo);
const suggestedReply = await ctx.services.ai.generateReply(
intelligence.messages,
conversation.username,
@@ -202,11 +260,12 @@ export function createConversationsRouter(ctx: WebContext): Router {
return;
}
try {
const conversation = await findConversation(decodeURIComponent(req.params.id));
if (!conversation) {
const resolved = await findConversation(decodeURIComponent(req.params.id));
if (!resolved) {
res.status(404).json({ error: 'Conversation not found' });
return;
}
const conversation = resolved.conversation;
if (!conversation.messages || conversation.messages.length === 0) {
res.status(400).json({ error: 'No messages in conversation to generate from' });
return;
@@ -216,7 +275,7 @@ export function createConversationsRouter(ctx: WebContext): Router {
return;
}
const intelligence = getConversationIntelligence(conversation, 'media-ask');
const intelligence = getConversationIntelligence(conversation, resolved.runtime?.resources.conversationStateRepo, 'media-ask');
if (intelligence.intent.kind !== 'media-ask') {
res.status(409).json({ error: intelligence.intent.reason, intent: intelligence.intent, conversationState: intelligence.state });
return;
@@ -246,17 +305,20 @@ export function createConversationsRouter(ctx: WebContext): Router {
return;
}
const conversation = await findConversation(decodeURIComponent(req.params.id));
if (!conversation) {
const resolved = await findConversation(decodeURIComponent(req.params.id));
if (!resolved) {
res.status(404).json({ error: 'Conversation not found' });
return;
}
const conversation = resolved.conversation;
const chatService = resolved.runtime?.resources.chatService ?? ctx.services.chat;
const stateRepo = resolved.runtime?.resources.conversationStateRepo ?? ctx.services.conversationState;
if (!conversation.url) {
res.status(400).json({ error: 'No chat URL for this conversation' });
return;
}
const success = await ctx.services.chat.sendChatMessage(conversation.url, message.trim());
const success = await chatService.sendChatMessage(conversation.url, message.trim());
if (success) {
if (conversation.roomId) {
const currentMessages = (conversation.messages ?? []).map((item) => ({
@@ -264,18 +326,26 @@ export function createConversationsRouter(ctx: WebContext): Router {
text: item.text,
isSelf: item.isSelf,
}));
const saved = ctx.services.conversationState?.get(conversation.roomId);
const saved = stateRepo?.get(conversation.roomId);
const updatedState = mergeConversationState(
saved,
buildConversationState([...currentMessages, { sender: 'Me', text: message.trim(), isSelf: true }]),
);
ctx.services.conversationState?.save(conversation.roomId, updatedState);
stateRepo?.save(conversation.roomId, updatedState);
}
if (resolved.runtime) {
resolved.runtime.resources.activityLogRepo.insert({
timestamp: Date.now(), type: 'sent',
message: `Sent chat reply to u/${conversation.username}`,
recipient: conversation.username,
});
} else {
ctx.activity.push('sent', `Sent chat reply to u/${conversation.username}`, conversation.username);
}
ctx.activity.push('sent', `Sent chat reply to u/${conversation.username}`, conversation.username);
ctx.broadcaster.push('info', `Chat reply sent to u/${conversation.username}`);
res.json({ success: true });
ctx.services.chat.refreshTopChats(5).catch((err) =>
chatService.refreshTopChats(5).catch((err) =>
ctx.logger.error('Failed to refresh chats after send-reply:', err),
);
} else {
@@ -297,7 +367,11 @@ export function createConversationsRouter(ctx: WebContext): Router {
return;
}
try {
const { username, message } = req.body ?? {};
const { accountId, username, message } = req.body ?? {};
if (!accountId || typeof accountId !== 'string') {
res.status(400).json({ error: 'accountId is required' });
return;
}
if (!username || typeof username !== 'string') {
res.status(400).json({ error: 'username is required' });
return;
@@ -306,20 +380,29 @@ export function createConversationsRouter(ctx: WebContext): Router {
res.status(400).json({ error: 'message is required' });
return;
}
const chat = await ctx.services.chat.startConversation(username, message.trim());
const runtime = ctx.services.accountRuntimes?.get(accountId);
if (!runtime || runtime.snapshot().state !== 'running') {
res.status(409).json({ error: 'Selected Reddit account is not running' });
return;
}
const chat = await runtime.resources.chatService.startConversation(username, message.trim());
if (!chat) {
res.status(500).json({ error: 'Failed to start conversation (unknown user or send failed)' });
return;
}
ctx.activity.push('sent', `Started chat with u/${chat.username}`, chat.username);
runtime.resources.activityLogRepo.insert({
timestamp: Date.now(), type: 'sent',
message: `Started chat with u/${chat.username}`,
recipient: chat.username,
});
if (chat.roomId) {
ctx.services.conversationState?.save(
runtime.resources.conversationStateRepo.save(
chat.roomId,
buildConversationState(chat.messages ?? [{ sender: 'Me', text: message.trim(), isSelf: true }]),
);
}
ctx.broadcaster.push('info', `Started new chat with u/${chat.username}`);
res.json({ success: true, chat });
res.json({ success: true, chat: { ...chat, owner: { accountId, redditUsername: runtime.session.redditUsername } } });
} catch (error) {
ctx.logger.error('Failed to start conversation:', error);
res.status(500).json({ error: String(error) });
src/web/routes/login.ts
+55 -4
View File
@@ -46,6 +46,18 @@ export function createLoginRouter(ctx: WebContext): Router {
const router = Router();
router.get('/login/status', async (_req, res) => {
const runningAccounts = ctx.services.redditAccounts?.list().filter((account) => account.isRunning) ?? [];
if (runningAccounts.length > 0) {
const only = runningAccounts.length === 1 ? runningAccounts[0] : null;
res.json({
ready: true,
loggedIn: true,
status: 'authenticated',
accountId: only?.accountId ?? null,
username: only?.redditUsername ?? null,
});
return;
}
const browser = ctx.services.browser;
if (!browser) {
res.status(503).json({ ready: false, error: 'browser not initialised' });
@@ -57,9 +69,16 @@ export function createLoginRouter(ctx: WebContext): Router {
? await accountAuth.ensureFullyAuthenticated()
: await browser.isSessionValid();
if (fullyAuthenticated) {
const username = accountAuth?.getSnapshot().redditUsername
const snapshot = accountAuth?.getSnapshot();
const username = snapshot?.redditUsername
?? await browser.getAuthenticatedUsername();
res.json({ ready: true, loggedIn: true, status: 'authenticated', username });
res.json({
ready: true,
loggedIn: true,
status: 'authenticated',
...(snapshot?.accountId ? { accountId: snapshot.accountId } : {}),
username,
});
return;
}
@@ -97,9 +116,16 @@ export function createLoginRouter(ctx: WebContext): Router {
}
try {
await importer.import(cookies.map(toPlaywrightCookie), payload.data.matrix);
const accounts = ctx.services.redditAccounts;
if (!accounts) {
await importer.import(cookies.map(toPlaywrightCookie), payload.data.matrix);
ctx.logger.info(`Imported ${cookies.length} Reddit session cookies and verified the browser session`);
res.json({ ok: true, status: 'authenticated' });
return;
}
const account = await accounts.importSession(cookies.map(toPlaywrightCookie), payload.data.matrix);
ctx.logger.info(`Imported ${cookies.length} Reddit session cookies and verified the browser session`);
res.json({ ok: true, status: 'authenticated' });
res.json({ ok: true, status: 'authenticated', account });
} catch (error) {
ctx.logger.error('Failed to import Reddit session cookies:', error);
const message = error instanceof Error ? error.message : 'Could not import Reddit session';
@@ -107,5 +133,30 @@ export function createLoginRouter(ctx: WebContext): Router {
}
});
router.get('/accounts', (_req, res) => {
const accounts = ctx.services.redditAccounts;
if (!accounts) {
res.status(503).json({ error: 'Reddit account service not initialised' });
return;
}
res.json(accounts.list());
});
router.post('/accounts/:accountId/activate', async (req, res) => {
const accounts = ctx.services.redditAccounts;
if (!accounts) {
res.status(503).json({ ok: false, error: 'Reddit account service not initialised' });
return;
}
try {
const account = await accounts.activate(decodeURIComponent(req.params.accountId));
ctx.logger.info(`Activated Reddit account u/${account.redditUsername}`);
res.json({ ok: true, account });
} catch (error) {
ctx.logger.error('Failed to activate Reddit account:', error);
res.status(401).json({ ok: false, error: error instanceof Error ? error.message : 'Could not activate Reddit account' });
}
});
return router;
}
src/web/routes/media.ts
+56 -7
View File
@@ -11,19 +11,27 @@ export function createMediaRouter(ctx: WebContext): Router {
return;
}
try {
const items = await ctx.services.media.getMediaItems();
const runtimes = ctx.services.accountRuntimes?.list().filter((runtime) => runtime.state === 'running') ?? [];
const items = runtimes.length > 0
? (await Promise.all(runtimes.map((snapshot) =>
ctx.services.accountRuntimes!.get(snapshot.accountId)!.resources.mediaService.getMediaItems(),
))).flat()
: await ctx.services.media.getMediaItems();
// Build a roomId → current chat lookup so we can override stale
// stored usernames (e.g. items captured before username enrichment ran)
// and surface the recipient even for rooms that were originally
// ingested as `unknown`.
const liveChats = ctx.services.chat
? await ctx.services.chat.getCachedConversations()
: [];
const byRoom = new Map(liveChats.map((c) => [c.roomId ?? '', c] as const));
const liveChats = runtimes.length > 0
? (await Promise.all(runtimes.map((snapshot) =>
ctx.services.accountRuntimes!.get(snapshot.accountId)!.resources.chatService.getCachedConversations()
.then((chats) => chats.map((chat) => ({ ...chat, accountId: snapshot.accountId }))),
))).flat()
: (ctx.services.chat ? await ctx.services.chat.getCachedConversations() : []).map((chat) => ({ ...chat, accountId: '' }));
const byRoom = new Map(liveChats.map((c) => [`${c.accountId}:${c.roomId ?? ''}`, c] as const));
res.json(items.map((item) => {
const live = item.chatRoomId ? byRoom.get(item.chatRoomId) : undefined;
const live = item.chatRoomId ? byRoom.get(`${item.accountId ?? ''}:${item.chatRoomId}`) : undefined;
const username = resolveRedditUsername(
live?.username ?? item.chatUsername,
ctx.services.chat,
@@ -31,11 +39,15 @@ export function createMediaRouter(ctx: WebContext): Router {
const sender = /^(me|you)$/i.test(item.sender)
? 'You'
: resolveRedditUsername(item.sender, ctx.services.chat) ?? username ?? 'Unknown Reddit user';
const account = item.accountId
? ctx.services.redditAccounts?.get(item.accountId)
: null;
return {
...item,
chatUsername: username,
sender,
urlPath: `/media/${encodeURIComponent(item.relativePath)}`,
urlPath: `/media/${item.relativePath.split(/[\\/]/).map(encodeURIComponent).join('/')}`,
owner: account ? { accountId: account.accountId, redditUsername: account.redditUsername } : undefined,
};
}));
} catch (error) {
@@ -45,6 +57,21 @@ export function createMediaRouter(ctx: WebContext): Router {
});
router.get('/media/sync', (_req, res) => {
const runtimes = ctx.services.accountRuntimes?.list().filter((runtime) => runtime.state === 'running') ?? [];
if (runtimes.length > 0) {
const statuses = runtimes.map((snapshot) =>
ctx.services.accountRuntimes!.get(snapshot.accountId)!.resources.mediaSyncService.getStatus(),
);
res.json({
isSyncing: statuses.some((status) => status.isSyncing),
lastSyncAt: newest(statuses.map((status) => status.lastSyncAt)),
lastSyncDurationMs: newest(statuses.map((status) => status.lastSyncDurationMs)),
lastSyncChats: statuses.reduce((total, status) => total + (status.lastSyncChats ?? 0), 0) || null,
lastError: statuses.find((status) => status.lastError)?.lastError ?? null,
nextSyncAt: newest(statuses.map((status) => status.nextSyncAt)),
});
return;
}
if (!ctx.services.mediaSync) {
res.status(503).json({ error: 'Media sync service not initialized' });
return;
@@ -53,6 +80,15 @@ export function createMediaRouter(ctx: WebContext): Router {
});
router.post('/media/sync', async (req, res) => {
const runtimes = ctx.services.accountRuntimes?.list().filter((runtime) => runtime.state === 'running') ?? [];
if (runtimes.length > 0) {
const full = req.body?.full === true;
const results = await Promise.all(runtimes.map((snapshot) =>
ctx.services.accountRuntimes!.get(snapshot.accountId)!.resources.mediaSyncService.triggerSync(full),
));
res.json({ success: results.some(Boolean), reason: results.some(Boolean) ? undefined : 'Sync already in progress' });
return;
}
if (!ctx.services.mediaSync) {
res.status(503).json({ error: 'Media sync service not initialized' });
return;
@@ -73,6 +109,14 @@ export function createMediaRouter(ctx: WebContext): Router {
});
router.post('/media/sync/full', async (_req, res) => {
const runtimes = ctx.services.accountRuntimes?.list().filter((runtime) => runtime.state === 'running') ?? [];
if (runtimes.length > 0) {
const results = await Promise.all(runtimes.map((snapshot) =>
ctx.services.accountRuntimes!.get(snapshot.accountId)!.resources.mediaSyncService.triggerSync(true),
));
res.json({ success: results.some(Boolean), reason: results.some(Boolean) ? undefined : 'Sync already in progress' });
return;
}
if (!ctx.services.mediaSync) {
res.status(503).json({ error: 'Media sync service not initialized' });
return;
@@ -93,3 +137,8 @@ export function createMediaRouter(ctx: WebContext): Router {
return router;
}
function newest(values: Array<number | null>): number | null {
const present = values.filter((value): value is number => value !== null);
return present.length > 0 ? Math.max(...present) : null;
}
src/web/routes/stats.ts
+57 -11
View File
@@ -1,31 +1,55 @@
import { Router } from 'express';
import type { WebContext } from '../context.js';
import type { BotStats } from '../types.js';
import type { ActivityItem, BotStats } from '../types.js';
import { presentActivityItem } from '../reddit-identities.js';
export function createStatsRouter(ctx: WebContext): Router {
const router = Router();
const recentActivity = (limit: number) => ctx.activity
.recent(limit)
.map((item) => presentActivityItem(item, ctx.services.chat));
const recentActivity = (limit: number) => {
const runtimes = runningRuntimes(ctx);
if (runtimes.length === 0) {
return ctx.activity.recent(limit).map((item) => presentActivityItem(item, ctx.services.chat));
}
return runtimes.flatMap((runtime) => runtime.resources.activityLogRepo.getRecent(limit).map((item) => ({
...presentActivityItem({
...item,
type: item.type as ActivityItem['type'],
recipient: item.recipient ?? undefined,
imageUrl: item.imageUrl ?? undefined,
postTitle: item.postTitle ?? undefined,
postUrl: item.postUrl ?? undefined,
}, runtime.resources.chatService),
owner: { accountId: runtime.session.accountId, redditUsername: runtime.session.redditUsername },
}))).sort((left, right) => right.timestamp - left.timestamp).slice(0, limit);
};
router.get('/stats', (_req, res) => {
const messages = ctx.repos.messages.getRecent(1000);
const runtimes = runningRuntimes(ctx);
const messages = runtimes.length > 0
? runtimes.flatMap((runtime) => runtime.resources.messagesRepo.getRecent(1000))
: ctx.repos.messages.getRecent(1000);
const totalSent = messages.length;
const totalResponses = messages.filter((m) => m.gotResponse).length;
const cooldown = ctx.services.chat?.getInvitationCooldown() ?? null;
const cooldown = runtimes.map((runtime) => runtime.resources.chatService.getInvitationCooldown()).find(Boolean)
?? ctx.services.chat?.getInvitationCooldown()
?? null;
const messagesThisHour = runtimes.length > 0
? runtimes.reduce((total, runtime) => total + runtime.resources.rateLimiter.getStats().messagesThisHour, 0)
: ctx.getMessagesThisHour();
const stats: BotStats = {
isRunning: ctx.getBotRunning(),
dryRunMode: ctx.config.browser.dryRun,
totalProcessed: ctx.activity.totalProcessed(),
totalProcessed: runtimes.length > 0
? runtimes.reduce((total, runtime) => total + runtime.resources.activityLogRepo.countSentSkipped(), 0)
: ctx.activity.totalProcessed(),
totalMessagesSent: totalSent,
totalResponses,
responseRate: totalSent > 0 ? Math.round((totalResponses / totalSent) * 100) : 0,
messagesThisHour: ctx.getMessagesThisHour(),
maxMessagesPerHour: ctx.config.rateLimit.maxMessagesPerHour,
messagesThisHour,
maxMessagesPerHour: ctx.config.rateLimit.maxMessagesPerHour * Math.max(1, runtimes.length),
subreddits: ctx.config.reddit.subreddits.map((s) => ({ name: s.name, type: s.type })),
targetGenders: ctx.config.reddit.targetGenders,
recentActivity: recentActivity(10),
@@ -42,11 +66,24 @@ export function createStatsRouter(ctx: WebContext): Router {
});
router.get('/posts/by-author/:username', (req, res) => {
res.json(ctx.repos.posts.getByAuthor(req.params.username));
const runtimes = runningRuntimes(ctx);
res.json(runtimes.length > 0
? runtimes.flatMap((runtime) => runtime.resources.postsRepo.getByAuthor(req.params.username).map((post) => ({
...post,
owner: { accountId: runtime.session.accountId, redditUsername: runtime.session.redditUsername },
})))
: ctx.repos.posts.getByAuthor(req.params.username));
});
router.get('/messages', (_req, res) => {
res.json(ctx.repos.messages.getRecent(20));
const runtimes = runningRuntimes(ctx);
const messages = runtimes.length > 0
? runtimes.flatMap((runtime) => runtime.resources.messagesRepo.getRecent(20).map((message) => ({
...message,
owner: { accountId: runtime.session.accountId, redditUsername: runtime.session.redditUsername },
})))
: ctx.repos.messages.getRecent(20);
res.json(messages.sort((left, right) => right.sentAt - left.sentAt).slice(0, 20));
});
router.get('/successful', (_req, res) => {
@@ -55,3 +92,12 @@ export function createStatsRouter(ctx: WebContext): Router {
return router;
}
function runningRuntimes(ctx: WebContext) {
return (ctx.services.accountRuntimes?.list() ?? [])
.filter((snapshot) => snapshot.state === 'running')
.flatMap((snapshot) => {
const runtime = ctx.services.accountRuntimes?.get(snapshot.accountId);
return runtime ? [runtime] : [];
});
}
src/web/server.ts
+10
View File
@@ -22,6 +22,8 @@ import type { BrowserClient } from '../browser/client.js';
import type { RedditSessionImporter } from '../services/reddit-session-importer.js';
import type { AccountAuthContext } from '../services/account-auth-context.js';
import type { ConfigStore } from '../config/persistence.js';
import type { RedditAccountsService } from '../services/reddit-accounts.js';
import type { RedditAccountRuntimeManager } from '../services/reddit-account-runtime.js';
import { ActivityLog } from './activity-log.js';
import { LogBroadcaster } from './log-broadcaster.js';
import type { WebContext } from './context.js';
@@ -219,6 +221,14 @@ export class WebServer {
this.services.configStore = configStore;
}
setRedditAccountsService(accounts: RedditAccountsService): void {
this.services.redditAccounts = accounts;
}
setAccountRuntimeManager(runtimes: RedditAccountRuntimeManager): void {
this.services.accountRuntimes = runtimes;
}
logActivity(
type: ActivityItem['type'],
message: string,
src/web/types.ts
+1
View File
@@ -26,6 +26,7 @@ export interface ActivityItem {
imageUrl?: string;
postTitle?: string;
postUrl?: string;
owner?: { accountId: string; redditUsername: string };
}
export interface LogEntry {