Files
RedditBot/src/database/repositories.ts
T
jjbruntonandClaude Opus 4.7 1b6dfd67e3 Persist t2→username mappings in reddit_users table
Adds a durable mapping so the UI never has to show raw t2_xxx IDs or
MXIDs just because the in-memory cache is cold after a restart.
ChatService seeds from the table on construction and writes through
every resolution path; cached chats whose username is still t2_xxx
get upgraded from the DB on loadCache.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-22 20:34:06 +01:00

649 lines
20 KiB
TypeScript

import type Database from 'better-sqlite3';
export interface ProcessedPostRecord {
id: string;
subreddit: string;
author: string;
title: string;
imageUrl: string | null;
imageAnalysis: string | null;
generatedMessage: string | null;
processedAt: number;
messageSent: boolean;
error: string | null;
}
export interface SentMessageRecord {
id: number;
postId: string | null;
recipient: string;
subject: string;
message: string;
sentAt: number;
gotResponse: boolean;
responseCheckedAt: number | null;
strategyId: string | null;
chatRoomId: string | null;
subreddit?: string;
postTitle?: string;
postImageUrl?: string;
postImageAnalysis?: string;
}
export interface SuccessfulMessageRecord {
id: number;
sentMessageId: number;
originalMessage: string;
imageAnalysis: string | null;
postTitle: string | null;
responseSnippet: string | null;
subreddit: string | null;
createdAt: number;
}
export interface StrategyStats {
strategyId: string;
totalSent: number;
totalResponses: number;
responseRate: number;
}
export interface SubredditStats {
subreddit: string;
totalSent: number;
totalResponses: number;
responseRate: number;
}
export interface DailyTrend {
date: string;
totalSent: number;
totalResponses: number;
responseRate: number;
}
export interface ActivityLogRecord {
timestamp: number;
type: string;
message: string;
recipient: string | null;
imageUrl: string | null;
}
export class ActivityLogRepository {
private insertStmt: Database.Statement;
private getRecentStmt: Database.Statement;
private pruneStmt: Database.Statement;
private countSentSkippedStmt: Database.Statement;
constructor(private db: Database.Database) {
this.insertStmt = db.prepare(`
INSERT INTO activity_log (timestamp, type, message, recipient, image_url)
VALUES (?, ?, ?, ?, ?)
`);
this.getRecentStmt = db.prepare(`
SELECT timestamp, type, message, recipient, image_url
FROM activity_log
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 ?)
`);
this.countSentSkippedStmt = db.prepare(`
SELECT COUNT(*) AS count FROM activity_log WHERE type IN ('sent', 'skipped')
`);
}
insert(item: {
timestamp: number;
type: string;
message: string;
recipient?: string;
imageUrl?: string;
}): void {
this.insertStmt.run(
item.timestamp,
item.type,
item.message,
item.recipient ?? null,
item.imageUrl ?? null,
);
}
getRecent(limit: number): ActivityLogRecord[] {
const rows = this.getRecentStmt.all(limit) as Record<string, unknown>[];
return rows.map((row) => ({
timestamp: row.timestamp as number,
type: row.type as string,
message: row.message as string,
recipient: (row.recipient as string) ?? null,
imageUrl: (row.image_url as string) ?? null,
}));
}
prune(keep: number): void {
this.pruneStmt.run(keep);
}
countSentSkipped(): number {
const row = this.countSentSkippedStmt.get() as { count: number };
return row.count;
}
}
export class PostsRepository {
private insertStmt: Database.Statement;
private existsStmt: Database.Statement;
private markSentStmt: Database.Statement;
private updateAnalysisStmt: Database.Statement;
private setErrorStmt: Database.Statement;
private getByIdStmt: Database.Statement;
private getByAuthorStmt: Database.Statement;
constructor(private db: Database.Database) {
this.insertStmt = db.prepare(`
INSERT INTO processed_posts (id, subreddit, author, title, image_url, processed_at)
VALUES (?, ?, ?, ?, ?, ?)
`);
this.existsStmt = db.prepare(`SELECT 1 FROM processed_posts WHERE id = ?`);
this.markSentStmt = db.prepare(`UPDATE processed_posts SET message_sent = 1 WHERE id = ?`);
this.updateAnalysisStmt = db.prepare(`
UPDATE processed_posts
SET image_analysis = ?, generated_message = ?
WHERE id = ?
`);
this.setErrorStmt = db.prepare(`UPDATE processed_posts SET error = ? WHERE id = ?`);
this.getByIdStmt = db.prepare(`SELECT * FROM processed_posts WHERE id = ?`);
this.getByAuthorStmt = db.prepare(`
SELECT * FROM processed_posts WHERE author = ? COLLATE NOCASE ORDER BY processed_at DESC LIMIT 10
`);
}
isProcessed(postId: string): boolean {
return this.existsStmt.get(postId) !== undefined;
}
insert(post: {
id: string;
subreddit: string;
author: string;
title: string;
imageUrl: string | null;
processedAt: number;
}): void {
this.insertStmt.run(
post.id,
post.subreddit,
post.author,
post.title,
post.imageUrl,
post.processedAt
);
}
updateAnalysis(postId: string, analysis: string, message: string): void {
this.updateAnalysisStmt.run(analysis, message, postId);
}
markMessageSent(postId: string): void {
this.markSentStmt.run(postId);
}
setError(postId: string, error: string): void {
this.setErrorStmt.run(error, postId);
}
getById(postId: string): ProcessedPostRecord | undefined {
const row = this.getByIdStmt.get(postId) as Record<string, unknown> | undefined;
if (!row) return undefined;
return {
id: row.id as string,
subreddit: row.subreddit as string,
author: row.author as string,
title: row.title as string,
imageUrl: row.image_url as string | null,
imageAnalysis: row.image_analysis as string | null,
generatedMessage: row.generated_message as string | null,
processedAt: row.processed_at as number,
messageSent: Boolean(row.message_sent),
error: row.error as string | null,
};
}
getByAuthor(author: string): ProcessedPostRecord[] {
const rows = this.getByAuthorStmt.all(author) as Record<string, unknown>[];
return rows.map((row) => ({
id: row.id as string,
subreddit: row.subreddit as string,
author: row.author as string,
title: row.title as string,
imageUrl: row.image_url as string | null,
imageAnalysis: row.image_analysis as string | null,
generatedMessage: row.generated_message as string | null,
processedAt: row.processed_at as number,
messageSent: Boolean(row.message_sent),
error: row.error as string | null,
}));
}
}
export class MessagesRepository {
private insertStmt: Database.Statement;
private insertChatStmt: Database.Statement;
private getByPostIdStmt: Database.Statement;
private getByChatRoomIdStmt: Database.Statement;
private getLatestByRecipientStmt: Database.Statement;
private getRecentStmt: Database.Statement;
private getUncheckedStmt: Database.Statement;
private markResponseStmt: Database.Statement;
private markCheckedStmt: Database.Statement;
private hasSentToStmt: Database.Statement;
constructor(private db: Database.Database) {
this.insertStmt = db.prepare(`
INSERT INTO sent_messages (post_id, recipient, subject, message, sent_at, strategy_id)
VALUES (?, ?, ?, ?, ?, ?)
`);
this.insertChatStmt = db.prepare(`
INSERT INTO sent_messages (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.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 = ?
LIMIT 1
`);
this.getByPostIdStmt = db.prepare(`SELECT * FROM sent_messages WHERE post_id = ?`);
this.getLatestByRecipientStmt = db.prepare(`
SELECT sm.*, pp.subreddit, pp.title AS post_title, 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
ORDER BY sm.sent_at DESC
LIMIT 1
`);
this.getRecentStmt = db.prepare(`
SELECT sm.*, pp.subreddit, pp.title AS post_title, 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
ORDER BY sm.sent_at DESC
LIMIT ?
`);
this.getUncheckedStmt = db.prepare(`
SELECT * FROM sent_messages
WHERE got_response = 0
AND (response_checked_at IS NULL OR response_checked_at < ?)
AND sent_at < ?
ORDER BY sent_at ASC
LIMIT ?
`);
this.markResponseStmt = db.prepare(`
UPDATE sent_messages SET got_response = 1, response_checked_at = ? WHERE id = ?
`);
this.markCheckedStmt = db.prepare(`
UPDATE sent_messages SET response_checked_at = ? WHERE id = ?
`);
this.hasSentToStmt = db.prepare(`
SELECT 1 FROM sent_messages WHERE recipient = ? LIMIT 1
`);
}
insert(message: {
postId: string;
recipient: string;
subject: string;
message: string;
sentAt: number;
strategyId?: string | null;
}): number {
const result = this.insertStmt.run(
message.postId,
message.recipient,
message.subject,
message.message,
message.sentAt,
message.strategyId || null
);
return result.lastInsertRowid as number;
}
insertChat(message: {
recipient: string;
subject: string;
message: string;
sentAt: number;
chatRoomId: string;
}): number {
const result = this.insertChatStmt.run(
message.recipient,
message.subject,
message.message,
message.sentAt,
message.chatRoomId,
);
return result.lastInsertRowid as number;
}
getByPostId(postId: string): SentMessageRecord | undefined {
const row = this.getByPostIdStmt.get(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;
if (!row) return undefined;
return this.mapRow(row);
}
getLatestByRecipient(recipient: string): SentMessageRecord | undefined {
const row = this.getLatestByRecipientStmt.get(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>[];
return rows.map((row) => this.mapRow(row));
}
getUncheckedMessages(minAgeMs: number, recheckAfterMs: number, limit: number = 10): SentMessageRecord[] {
const now = Date.now();
const minSentAt = now - minAgeMs;
const recheckBefore = now - recheckAfterMs;
const rows = this.getUncheckedStmt.all(recheckBefore, minSentAt, limit) as Record<string, unknown>[];
return rows.map((row) => this.mapRow(row));
}
markGotResponse(messageId: number): void {
this.markResponseStmt.run(Date.now(), messageId);
}
markChecked(messageId: number): void {
this.markCheckedStmt.run(Date.now(), messageId);
}
hasSentTo(recipient: string): boolean {
return this.hasSentToStmt.get(recipient) !== undefined;
}
getStrategyStats(): StrategyStats[] {
const rows = this.db.prepare(`
SELECT
strategy_id,
COUNT(*) as total_sent,
SUM(CASE WHEN got_response = 1 THEN 1 ELSE 0 END) as total_responses
FROM sent_messages
WHERE strategy_id IS NOT NULL
GROUP BY strategy_id
`).all() as Record<string, unknown>[];
return rows.map(row => ({
strategyId: row.strategy_id as string,
totalSent: row.total_sent as number,
totalResponses: row.total_responses as number,
responseRate: (row.total_sent as number) > 0
? Math.round(((row.total_responses as number) / (row.total_sent as number)) * 100)
: 0,
}));
}
getSubredditStats(): SubredditStats[] {
const rows = this.db.prepare(`
SELECT
pp.subreddit,
COUNT(*) as total_sent,
SUM(CASE WHEN sm.got_response = 1 THEN 1 ELSE 0 END) as total_responses
FROM sent_messages sm
JOIN processed_posts pp ON sm.post_id = pp.id
GROUP BY pp.subreddit
ORDER BY total_sent DESC
`).all() as Record<string, unknown>[];
return rows.map(row => ({
subreddit: row.subreddit as string,
totalSent: row.total_sent as number,
totalResponses: row.total_responses as number,
responseRate: (row.total_sent as number) > 0
? Math.round(((row.total_responses as number) / (row.total_sent as number)) * 100)
: 0,
}));
}
getResponseRateTrend(days: number = 30): DailyTrend[] {
const cutoff = Date.now() - (days * 24 * 60 * 60 * 1000);
const rows = this.db.prepare(`
SELECT
date(sent_at / 1000, 'unixepoch') as day,
COUNT(*) as total_sent,
SUM(CASE WHEN got_response = 1 THEN 1 ELSE 0 END) as total_responses
FROM sent_messages
WHERE sent_at > ?
GROUP BY day
ORDER BY day ASC
`).all(cutoff) as Record<string, unknown>[];
return rows.map(row => ({
date: row.day as string,
totalSent: row.total_sent as number,
totalResponses: row.total_responses as number,
responseRate: (row.total_sent as number) > 0
? Math.round(((row.total_responses as number) / (row.total_sent as number)) * 100)
: 0,
}));
}
private mapRow(row: Record<string, unknown>): SentMessageRecord {
return {
id: row.id as number,
postId: (row.post_id as string) ?? null,
recipient: row.recipient as string,
subject: row.subject as string,
message: row.message as string,
sentAt: row.sent_at as number,
gotResponse: Boolean(row.got_response),
responseCheckedAt: row.response_checked_at as number | null,
strategyId: (row.strategy_id as string) || null,
chatRoomId: (row.chat_room_id as string) ?? null,
subreddit: (row.subreddit as string) || undefined,
postTitle: (row.post_title as string) || undefined,
postImageUrl: (row.post_image_url as string) || undefined,
postImageAnalysis: (row.post_image_analysis as string) || undefined,
};
}
}
export class SuccessfulMessagesRepository {
private insertStmt: Database.Statement;
private getRecentStmt: Database.Statement;
private getBySubredditStmt: Database.Statement;
private countStmt: Database.Statement;
private existsByMessageStmt: Database.Statement;
constructor(private db: Database.Database) {
this.insertStmt = db.prepare(`
INSERT INTO successful_messages (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 ?
`);
// Get examples prioritizing same subreddit, then falling back to others
this.getBySubredditStmt = db.prepare(`
SELECT *,
CASE WHEN subreddit = ? THEN 1 ELSE 0 END as subreddit_match
FROM successful_messages
ORDER BY subreddit_match DESC, created_at DESC
LIMIT ?
`);
this.countStmt = db.prepare(`SELECT COUNT(*) as count FROM successful_messages`);
this.existsByMessageStmt = db.prepare(`
SELECT 1 FROM successful_messages WHERE original_message = ? LIMIT 1
`);
}
insert(data: {
sentMessageId: number | null;
originalMessage: string;
imageAnalysis: string | null;
postTitle: string | null;
responseSnippet: string | null;
subreddit: string | null;
}): void {
this.insertStmt.run(
data.sentMessageId,
data.originalMessage,
data.imageAnalysis,
data.postTitle,
data.responseSnippet,
data.subreddit,
Date.now()
);
}
existsByMessage(originalMessage: string): boolean {
return this.existsByMessageStmt.get(originalMessage) !== undefined;
}
getRecent(limit: number = 5): SuccessfulMessageRecord[] {
const rows = this.getRecentStmt.all(limit) as Record<string, unknown>[];
return rows.map((row) => this.mapRow(row));
}
getWeightedExamples(subreddit: string | null, limit: number = 3): SuccessfulMessageRecord[] {
if (subreddit) {
const rows = this.getBySubredditStmt.all(subreddit, 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 };
return result.count;
}
private mapRow(row: Record<string, unknown>): SuccessfulMessageRecord {
return {
id: row.id as number,
sentMessageId: row.sent_message_id as number,
originalMessage: row.original_message as string,
imageAnalysis: row.image_analysis as string | null,
postTitle: row.post_title as string | null,
responseSnippet: row.response_snippet as string | null,
subreddit: row.subreddit as string | null,
createdAt: row.created_at as number,
};
}
}
export interface RedditUserRecord {
t2Id: string;
username: string;
mxid: string | null;
updatedAt: number;
}
/**
* Durable mapping of Reddit account IDs (`t2_xxx`) to current usernames.
* Populated as the chat sync resolves IDs via Reddit's alfred/user_data
* endpoints; read back on startup so we never have to show raw `t2_xxx`
* or MXIDs in the UI just because the in-memory cache is cold.
*/
export class RedditUsersRepository {
private upsertStmt: Database.Statement;
private getByT2Stmt: Database.Statement;
private getByUsernameStmt: Database.Statement;
private getAllStmt: Database.Statement;
constructor(private db: Database.Database) {
this.upsertStmt = db.prepare(`
INSERT INTO reddit_users (t2_id, username, mxid, updated_at)
VALUES (@t2Id, @username, @mxid, @updatedAt)
ON CONFLICT(t2_id) DO UPDATE SET
username = excluded.username,
mxid = COALESCE(excluded.mxid, reddit_users.mxid),
updated_at = excluded.updated_at
WHERE excluded.username != ''
`);
this.getByT2Stmt = db.prepare(`SELECT * FROM reddit_users WHERE t2_id = ?`);
this.getByUsernameStmt = db.prepare(`SELECT * FROM reddit_users WHERE username = ? COLLATE NOCASE`);
this.getAllStmt = db.prepare(`SELECT * FROM reddit_users`);
}
upsert(record: { t2Id: string; username: string; mxid?: string | null }): void {
if (!record.t2Id || !record.username) return;
if (record.username.startsWith('t2_') || record.username.startsWith('@')) return;
const t2Id = record.t2Id.startsWith('t2_') ? record.t2Id : `t2_${record.t2Id}`;
this.upsertStmt.run({
t2Id,
username: record.username,
mxid: record.mxid ?? null,
updatedAt: Date.now(),
});
}
upsertMany(records: Array<{ t2Id: string; username: string; mxid?: string | null }>): void {
const tx = this.db.transaction((items: typeof records) => {
for (const r of items) this.upsert(r);
});
tx(records);
}
getByT2(t2Id: string): RedditUserRecord | null {
const normalized = t2Id.startsWith('t2_') ? t2Id : `t2_${t2Id}`;
const row = this.getByT2Stmt.get(normalized) as Record<string, unknown> | undefined;
return row ? this.mapRow(row) : null;
}
getByUsername(username: string): RedditUserRecord | null {
const row = this.getByUsernameStmt.get(username) as Record<string, unknown> | undefined;
return row ? this.mapRow(row) : null;
}
/** Bulk load — used to seed the in-memory cache on startup. */
getAll(): RedditUserRecord[] {
const rows = this.getAllStmt.all() as Record<string, unknown>[];
return rows.map((r) => this.mapRow(r));
}
private mapRow(row: Record<string, unknown>): RedditUserRecord {
return {
t2Id: row.t2_id as string,
username: row.username as string,
mxid: (row.mxid as string | null) ?? null,
updatedAt: row.updated_at as number,
};
}
}