761 lines
31 KiB
TypeScript
761 lines
31 KiB
TypeScript
import Database from 'better-sqlite3';
|
|
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:'
|
|
: path.resolve(process.cwd(), config.database.path);
|
|
const dbDir = path.dirname(dbPath);
|
|
|
|
// Ensure directory exists
|
|
if (!fs.existsSync(dbDir)) {
|
|
fs.mkdirSync(dbDir, { recursive: true });
|
|
}
|
|
|
|
const db = new Database(dbPath);
|
|
|
|
// Enable WAL mode for better concurrency
|
|
db.pragma('journal_mode = WAL');
|
|
db.pragma('foreign_keys = ON');
|
|
|
|
return db;
|
|
}
|
|
|
|
export function runMigrations(db: Database.Database): void {
|
|
const migrations: ReadonlyArray<() => void> = [
|
|
() => db.exec(`
|
|
CREATE TABLE IF NOT EXISTS processed_posts (
|
|
id TEXT PRIMARY KEY,
|
|
subreddit TEXT NOT NULL,
|
|
author TEXT NOT NULL,
|
|
title TEXT NOT NULL,
|
|
image_url TEXT,
|
|
image_analysis TEXT,
|
|
generated_message TEXT,
|
|
post_created_at INTEGER,
|
|
processed_at INTEGER NOT NULL,
|
|
message_sent INTEGER DEFAULT 0,
|
|
error TEXT
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS sent_messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
post_id TEXT NOT NULL,
|
|
recipient TEXT NOT NULL,
|
|
subject TEXT NOT NULL,
|
|
message TEXT NOT NULL,
|
|
sent_at INTEGER NOT NULL,
|
|
FOREIGN KEY (post_id) REFERENCES processed_posts(id)
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS successful_messages (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
sent_message_id INTEGER,
|
|
original_message TEXT NOT NULL,
|
|
image_analysis TEXT,
|
|
post_title TEXT,
|
|
response_snippet TEXT,
|
|
created_at INTEGER NOT NULL
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS activity_log (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
timestamp INTEGER NOT NULL,
|
|
type TEXT NOT NULL,
|
|
message TEXT NOT NULL,
|
|
recipient TEXT,
|
|
image_url TEXT
|
|
);
|
|
|
|
CREATE TABLE IF NOT EXISTS reddit_users (
|
|
t2_id TEXT PRIMARY KEY,
|
|
username TEXT NOT NULL,
|
|
mxid TEXT,
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_reddit_users_username
|
|
ON reddit_users(username);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_posts_subreddit
|
|
ON processed_posts(subreddit);
|
|
CREATE INDEX IF NOT EXISTS idx_posts_processed_at
|
|
ON processed_posts(processed_at);
|
|
CREATE INDEX IF NOT EXISTS idx_posts_message_sent
|
|
ON processed_posts(message_sent);
|
|
CREATE INDEX IF NOT EXISTS idx_activity_log_timestamp
|
|
ON activity_log(timestamp);
|
|
`),
|
|
|
|
() => {
|
|
addColumnIfNotExists(db, 'sent_messages', 'got_response', 'INTEGER DEFAULT 0');
|
|
addColumnIfNotExists(db, 'sent_messages', 'response_checked_at', 'INTEGER');
|
|
addColumnIfNotExists(db, 'sent_messages', 'strategy_id', 'TEXT');
|
|
addColumnIfNotExists(db, 'sent_messages', 'chat_room_id', 'TEXT');
|
|
db.exec(`
|
|
CREATE INDEX IF NOT EXISTS idx_sent_messages_response ON sent_messages(got_response);
|
|
CREATE INDEX IF NOT EXISTS idx_sent_messages_recipient ON sent_messages(recipient);
|
|
CREATE INDEX IF NOT EXISTS idx_sent_messages_chat_room ON sent_messages(chat_room_id);
|
|
`);
|
|
makeSentMessagesPostIdNullable(db);
|
|
},
|
|
() => {
|
|
addColumnIfNotExists(db, 'successful_messages', 'subreddit', 'TEXT');
|
|
makeSuccessfulMessageIdNullable(db);
|
|
},
|
|
() => db.exec(`
|
|
CREATE TABLE IF NOT EXISTS delivery_outbox (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
post_id TEXT NOT NULL UNIQUE,
|
|
recipient TEXT NOT NULL,
|
|
subject TEXT NOT NULL,
|
|
message TEXT NOT NULL,
|
|
strategy_id TEXT,
|
|
idempotency_key TEXT NOT NULL UNIQUE,
|
|
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,
|
|
FOREIGN KEY (post_id) REFERENCES processed_posts(id),
|
|
FOREIGN KEY (sent_message_id) REFERENCES sent_messages(id)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_delivery_outbox_claim
|
|
ON delivery_outbox(status, lease_expires_at, created_at);
|
|
`),
|
|
() => db.exec(`
|
|
CREATE TABLE IF NOT EXISTS conversation_states (
|
|
conversation_id TEXT PRIMARY KEY,
|
|
state_json TEXT NOT NULL,
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
`),
|
|
() => db.exec(`
|
|
CREATE TABLE IF NOT EXISTS conversation_outcomes (
|
|
sent_message_id INTEGER PRIMARY KEY,
|
|
outcome TEXT NOT NULL CHECK(outcome IN ('engaged', 'declined', 'boundary', 'upsell', 'unknown')),
|
|
response_snippet TEXT,
|
|
recorded_at INTEGER NOT NULL,
|
|
FOREIGN KEY (sent_message_id) REFERENCES sent_messages(id)
|
|
);
|
|
`),
|
|
() => {
|
|
addColumnIfNotExists(db, 'processed_posts', 'post_url', 'TEXT');
|
|
addColumnIfNotExists(db, 'activity_log', 'post_title', 'TEXT');
|
|
addColumnIfNotExists(db, 'activity_log', 'post_url', 'TEXT');
|
|
db.exec(`
|
|
UPDATE processed_posts
|
|
SET post_url = 'https://www.reddit.com/r/' || subreddit || '/comments/' || id || '/'
|
|
WHERE post_url IS NULL;
|
|
|
|
UPDATE activity_log
|
|
SET
|
|
post_title = (
|
|
SELECT pp.title
|
|
FROM sent_messages sm
|
|
JOIN processed_posts pp ON pp.id = sm.post_id
|
|
WHERE sm.recipient = activity_log.recipient COLLATE NOCASE
|
|
AND sm.sent_at <= activity_log.timestamp
|
|
ORDER BY sm.sent_at DESC
|
|
LIMIT 1
|
|
),
|
|
post_url = (
|
|
SELECT pp.post_url
|
|
FROM sent_messages sm
|
|
JOIN processed_posts pp ON pp.id = sm.post_id
|
|
WHERE sm.recipient = activity_log.recipient COLLATE NOCASE
|
|
AND sm.sent_at <= activity_log.timestamp
|
|
ORDER BY sm.sent_at DESC
|
|
LIMIT 1
|
|
),
|
|
image_url = COALESCE(image_url, (
|
|
SELECT pp.image_url
|
|
FROM sent_messages sm
|
|
JOIN processed_posts pp ON pp.id = sm.post_id
|
|
WHERE sm.recipient = activity_log.recipient COLLATE NOCASE
|
|
AND sm.sent_at <= activity_log.timestamp
|
|
ORDER BY sm.sent_at DESC
|
|
LIMIT 1
|
|
))
|
|
WHERE type = 'sent'
|
|
AND recipient IS NOT NULL
|
|
AND post_url IS NULL;
|
|
`);
|
|
},
|
|
() => db.exec(`
|
|
CREATE TABLE IF NOT EXISTS app_config (
|
|
id INTEGER PRIMARY KEY CHECK(id = 1),
|
|
config_json TEXT NOT NULL,
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
`),
|
|
() => {
|
|
addColumnIfNotExists(db, 'sent_messages', 'got_image', 'INTEGER NOT NULL DEFAULT 0');
|
|
db.exec(`
|
|
CREATE INDEX IF NOT EXISTS idx_sent_messages_image ON sent_messages(got_image);
|
|
`);
|
|
},
|
|
() => migrateAccountOwnership(db),
|
|
() => {
|
|
addColumnIfNotExists(db, 'sent_messages', 'experiment_id', 'TEXT');
|
|
addColumnIfNotExists(db, 'sent_messages', 'prompt_version', 'TEXT');
|
|
addColumnIfNotExists(db, 'sent_messages', 'model_id', 'TEXT');
|
|
addColumnIfNotExists(db, 'sent_messages', 'selection_reason', 'TEXT');
|
|
addColumnIfNotExists(db, 'sent_messages', 'source_type', 'TEXT');
|
|
db.exec(`
|
|
CREATE INDEX IF NOT EXISTS idx_sent_messages_experiment
|
|
ON sent_messages(account_id, experiment_id, strategy_id, sent_at);
|
|
CREATE TABLE IF NOT EXISTS recipient_suppressions (
|
|
recipient TEXT PRIMARY KEY COLLATE NOCASE,
|
|
reason TEXT NOT NULL,
|
|
created_at INTEGER NOT NULL
|
|
);
|
|
`);
|
|
},
|
|
() => db.exec(`
|
|
CREATE INDEX IF NOT EXISTS idx_reddit_conversations_global_username
|
|
ON reddit_conversations(username COLLATE NOCASE);
|
|
CREATE INDEX IF NOT EXISTS idx_sent_messages_global_recipient
|
|
ON sent_messages(recipient COLLATE NOCASE);
|
|
|
|
INSERT OR IGNORE INTO recipient_suppressions (recipient, reason, created_at)
|
|
SELECT recipient, 'sent-message-history', MIN(sent_at)
|
|
FROM sent_messages
|
|
WHERE trim(recipient) <> ''
|
|
GROUP BY recipient COLLATE NOCASE;
|
|
|
|
INSERT OR IGNORE INTO recipient_suppressions (recipient, reason, created_at)
|
|
SELECT rc.username, 'historical-conversation', MIN(rc.updated_at)
|
|
FROM reddit_conversations rc
|
|
WHERE trim(rc.username) <> ''
|
|
AND lower(trim(rc.username)) NOT IN ('unknown', '[deleted]')
|
|
AND lower(trim(rc.username)) NOT LIKE 't2\\_%' ESCAPE '\\'
|
|
AND trim(rc.username) NOT LIKE '@%'
|
|
GROUP BY rc.username COLLATE NOCASE;
|
|
|
|
CREATE TABLE conversation_automation_claims (
|
|
account_id TEXT NOT NULL,
|
|
conversation_id TEXT NOT NULL,
|
|
incoming_key TEXT NOT NULL,
|
|
status TEXT NOT NULL CHECK(status IN ('pending', 'sent')),
|
|
claimed_at INTEGER NOT NULL,
|
|
sent_at INTEGER,
|
|
PRIMARY KEY (account_id, conversation_id, incoming_key),
|
|
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id) ON DELETE CASCADE
|
|
);
|
|
CREATE INDEX idx_conversation_automation_claims_status
|
|
ON conversation_automation_claims(account_id, status, claimed_at);
|
|
`),
|
|
() => db.exec(`
|
|
CREATE TABLE IF NOT EXISTS ai_usage (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
account_id TEXT,
|
|
operation TEXT NOT NULL,
|
|
requested_model TEXT NOT NULL,
|
|
actual_model TEXT,
|
|
outcome TEXT NOT NULL CHECK(outcome IN ('success', 'refusal', 'empty', 'error')),
|
|
prompt_tokens INTEGER NOT NULL DEFAULT 0,
|
|
completion_tokens INTEGER NOT NULL DEFAULT 0,
|
|
total_tokens INTEGER NOT NULL DEFAULT 0,
|
|
cost_usd REAL,
|
|
latency_ms INTEGER NOT NULL,
|
|
error TEXT,
|
|
created_at INTEGER NOT NULL,
|
|
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id) ON DELETE SET NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_ai_usage_created_at ON ai_usage(created_at DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_ai_usage_model ON ai_usage(actual_model, requested_model, created_at DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_ai_usage_account ON ai_usage(account_id, created_at DESC);
|
|
`),
|
|
() => {
|
|
addColumnIfNotExists(db, 'processed_posts', 'post_created_at', 'INTEGER');
|
|
},
|
|
() => db.exec(`
|
|
CREATE TABLE IF NOT EXISTS candidate_queue (
|
|
post_id TEXT PRIMARY KEY,
|
|
subreddit TEXT NOT NULL,
|
|
author TEXT NOT NULL COLLATE NOCASE,
|
|
source_type TEXT NOT NULL CHECK(source_type IN ('image', 'chat')),
|
|
payload_json TEXT NOT NULL,
|
|
prepared_json TEXT,
|
|
post_created_at INTEGER NOT NULL,
|
|
priority REAL NOT NULL DEFAULT 1,
|
|
status TEXT NOT NULL CHECK(status IN (
|
|
'discovered', 'screening', 'ready', 'sending', 'retryable', 'sent', 'rejected'
|
|
)),
|
|
discovered_at INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL,
|
|
available_at INTEGER NOT NULL,
|
|
lease_owner TEXT,
|
|
lease_expires_at INTEGER,
|
|
assigned_account_id TEXT,
|
|
attempt_count INTEGER NOT NULL DEFAULT 0,
|
|
last_error TEXT,
|
|
FOREIGN KEY (assigned_account_id) REFERENCES reddit_accounts(account_id) ON DELETE SET NULL
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_candidate_queue_claim
|
|
ON candidate_queue(status, available_at, priority DESC, post_created_at DESC);
|
|
CREATE INDEX IF NOT EXISTS idx_candidate_queue_author
|
|
ON candidate_queue(author COLLATE NOCASE, status);
|
|
CREATE INDEX IF NOT EXISTS idx_candidate_queue_updated
|
|
ON candidate_queue(updated_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS runtime_state (
|
|
id INTEGER PRIMARY KEY CHECK(id = 1),
|
|
bot_running INTEGER NOT NULL CHECK(bot_running IN (0, 1)),
|
|
updated_at INTEGER NOT NULL
|
|
);
|
|
`),
|
|
() => {
|
|
db.exec(`
|
|
CREATE TABLE IF NOT EXISTS conversation_automation_claims (
|
|
account_id TEXT NOT NULL,
|
|
conversation_id TEXT NOT NULL,
|
|
incoming_key TEXT NOT NULL,
|
|
status TEXT NOT NULL CHECK(status IN ('pending', 'sent')),
|
|
claimed_at INTEGER NOT NULL,
|
|
sent_at INTEGER,
|
|
reply_text TEXT,
|
|
incoming_at INTEGER,
|
|
last_attempt_at INTEGER,
|
|
attempt_count INTEGER NOT NULL DEFAULT 0,
|
|
PRIMARY KEY (account_id, conversation_id, incoming_key),
|
|
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id) ON DELETE CASCADE
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_conversation_automation_claims_status
|
|
ON conversation_automation_claims(account_id, status, claimed_at);
|
|
`);
|
|
addColumnIfNotExists(db, 'conversation_automation_claims', 'reply_text', 'TEXT');
|
|
addColumnIfNotExists(db, 'conversation_automation_claims', 'incoming_at', 'INTEGER');
|
|
addColumnIfNotExists(db, 'conversation_automation_claims', 'last_attempt_at', 'INTEGER');
|
|
addColumnIfNotExists(db, 'conversation_automation_claims', 'attempt_count', 'INTEGER NOT NULL DEFAULT 0');
|
|
},
|
|
() => db.exec(`
|
|
CREATE TABLE conversation_automation_claims_with_superseded (
|
|
account_id TEXT NOT NULL,
|
|
conversation_id TEXT NOT NULL,
|
|
incoming_key TEXT NOT NULL,
|
|
status TEXT NOT NULL CHECK(status IN ('pending', 'sent', 'superseded')),
|
|
claimed_at INTEGER NOT NULL,
|
|
sent_at INTEGER,
|
|
superseded_at INTEGER,
|
|
reply_text TEXT,
|
|
incoming_at INTEGER,
|
|
last_attempt_at INTEGER,
|
|
attempt_count INTEGER NOT NULL DEFAULT 0,
|
|
PRIMARY KEY (account_id, conversation_id, incoming_key),
|
|
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id) ON DELETE CASCADE
|
|
);
|
|
INSERT INTO conversation_automation_claims_with_superseded (
|
|
account_id, conversation_id, incoming_key, status, claimed_at, sent_at,
|
|
reply_text, incoming_at, last_attempt_at, attempt_count
|
|
) SELECT
|
|
account_id, conversation_id, incoming_key, status, claimed_at, sent_at,
|
|
reply_text, incoming_at, last_attempt_at, attempt_count
|
|
FROM conversation_automation_claims;
|
|
DROP TABLE conversation_automation_claims;
|
|
ALTER TABLE conversation_automation_claims_with_superseded RENAME TO conversation_automation_claims;
|
|
CREATE INDEX idx_conversation_automation_claims_status
|
|
ON conversation_automation_claims(account_id, status, claimed_at);
|
|
`),
|
|
() => db.exec(`
|
|
CREATE TABLE IF NOT EXISTS media_library_assets (
|
|
asset_id TEXT PRIMARY KEY,
|
|
tags_json TEXT NOT NULL,
|
|
safety TEXT NOT NULL CHECK(safety IN ('sfw', 'nsfw', 'unknown')),
|
|
classification_status TEXT NOT NULL CHECK(classification_status IN ('ready', 'failed')),
|
|
description TEXT,
|
|
confidence REAL,
|
|
file_name TEXT NOT NULL,
|
|
relative_path TEXT NOT NULL UNIQUE,
|
|
mime_type TEXT NOT NULL,
|
|
size_bytes INTEGER NOT NULL,
|
|
width INTEGER NOT NULL,
|
|
height INTEGER NOT NULL,
|
|
sha256 TEXT NOT NULL UNIQUE,
|
|
enabled INTEGER NOT NULL DEFAULT 0 CHECK(enabled IN (0, 1)),
|
|
consent_confirmed_at INTEGER NOT NULL,
|
|
created_at INTEGER NOT NULL,
|
|
updated_at INTEGER NOT NULL,
|
|
deleted_at INTEGER
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_media_library_assets_enabled
|
|
ON media_library_assets(enabled, classification_status, created_at DESC);
|
|
|
|
CREATE TABLE IF NOT EXISTS media_library_sends (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
asset_id TEXT NOT NULL,
|
|
asset_sha256 TEXT NOT NULL,
|
|
account_id TEXT NOT NULL,
|
|
conversation_id TEXT NOT NULL,
|
|
incoming_key TEXT NOT NULL,
|
|
request_kind TEXT NOT NULL CHECK(request_kind IN ('generic', 'penis', 'chest', 'selfie', 'topless', 'feet')),
|
|
status TEXT NOT NULL CHECK(status IN ('pending', 'sent', 'failed', 'superseded')),
|
|
requested_at INTEGER NOT NULL,
|
|
sent_at INTEGER,
|
|
matrix_media_event_id TEXT,
|
|
matrix_text_event_id TEXT,
|
|
mxc_uri TEXT,
|
|
error TEXT,
|
|
UNIQUE(account_id, conversation_id, incoming_key),
|
|
UNIQUE(account_id, conversation_id, asset_id),
|
|
UNIQUE(account_id, conversation_id, asset_sha256),
|
|
FOREIGN KEY (asset_id) REFERENCES media_library_assets(asset_id),
|
|
FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id) ON DELETE CASCADE
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_media_library_sends_conversation
|
|
ON media_library_sends(account_id, conversation_id, requested_at);
|
|
`),
|
|
];
|
|
|
|
let currentVersion = db.pragma('user_version', { simple: true }) as number;
|
|
if (currentVersion > migrations.length) {
|
|
throw new Error(`Database schema version ${currentVersion} is newer than this application supports`);
|
|
}
|
|
|
|
for (let index = currentVersion; index < migrations.length; index += 1) {
|
|
const version = index + 1;
|
|
// 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);
|
|
if (!exists) {
|
|
db.exec(`ALTER TABLE ${table} ADD COLUMN ${column} ${definition}`);
|
|
}
|
|
}
|
|
|
|
function makeSentMessagesPostIdNullable(db: Database.Database): void {
|
|
const sentMsgColumns = db.prepare(`PRAGMA table_info(sent_messages)`).all() as { name: string; notnull: number }[];
|
|
const postIdCol = sentMsgColumns.find((c) => c.name === 'post_id');
|
|
if (postIdCol && postIdCol.notnull === 1) {
|
|
db.exec(`
|
|
CREATE TABLE sent_messages_new (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
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,
|
|
response_checked_at INTEGER,
|
|
strategy_id TEXT,
|
|
chat_room_id TEXT,
|
|
FOREIGN KEY (post_id) REFERENCES processed_posts(id)
|
|
);
|
|
|
|
INSERT INTO sent_messages_new (id, post_id, recipient, subject, message, sent_at, got_response, response_checked_at, strategy_id, chat_room_id)
|
|
SELECT id, post_id, recipient, subject, message, sent_at, got_response, response_checked_at, strategy_id, chat_room_id
|
|
FROM sent_messages;
|
|
|
|
DROP TABLE sent_messages;
|
|
ALTER TABLE sent_messages_new RENAME TO sent_messages;
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_sent_messages_response ON sent_messages(got_response);
|
|
CREATE INDEX IF NOT EXISTS idx_sent_messages_recipient ON sent_messages(recipient);
|
|
CREATE INDEX IF NOT EXISTS idx_sent_messages_chat_room ON sent_messages(chat_room_id);
|
|
`);
|
|
}
|
|
}
|
|
|
|
function makeSuccessfulMessageIdNullable(db: Database.Database): void {
|
|
const successfulMsgColumns = db.prepare(`PRAGMA table_info(successful_messages)`).all() as { name: string; notnull: number }[];
|
|
const sentMsgIdCol = successfulMsgColumns.find((c) => c.name === 'sent_message_id');
|
|
|
|
if (sentMsgIdCol && sentMsgIdCol.notnull === 1) {
|
|
// Column is currently NOT NULL, we need to migrate
|
|
db.exec(`
|
|
-- Create new table with nullable sent_message_id
|
|
CREATE TABLE successful_messages_new (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
sent_message_id INTEGER,
|
|
original_message TEXT NOT NULL,
|
|
image_analysis TEXT,
|
|
post_title TEXT,
|
|
response_snippet TEXT,
|
|
created_at INTEGER NOT NULL
|
|
);
|
|
|
|
-- Copy existing data
|
|
INSERT INTO successful_messages_new
|
|
SELECT id, sent_message_id, original_message, image_analysis, post_title, response_snippet, created_at
|
|
FROM successful_messages;
|
|
|
|
-- Drop old table
|
|
DROP TABLE successful_messages;
|
|
|
|
-- Rename new table
|
|
ALTER TABLE successful_messages_new RENAME TO successful_messages;
|
|
`);
|
|
}
|
|
}
|