Files
linki/lib/db.ts
T
2026-08-01 14:01:17 +02:00

884 lines
40 KiB
TypeScript

import Database from "better-sqlite3";
import path from "path";
import { randomUUID } from "crypto";
import { scheduleUpdateCheck } from "@/lib/update-check";
import { encryptSecret, isEncrypted } from "@/lib/crypto";
const DB_PATH = process.env.LINKI_DB_PATH || path.join(process.cwd(), "linki.db");
let db: Database.Database;
export function getDb(): Database.Database {
if (!db) {
db = new Database(DB_PATH);
db.pragma("journal_mode = WAL");
db.pragma("foreign_keys = ON");
initDb(db);
runMigrations(db);
scheduleUpdateCheck();
}
return db;
}
function runParallelTracksMigration(db: Database.Database) {
// This backfill reads the legacy run_profiles.state column. If that column no longer
// exists, dropDeprecatedRunProfileColumns has already run (a prior startup) and this
// migration is moot — skip, otherwise the SELECT rp.state below throws "no such column".
// (Fresh DBs hit this: state is created, dropped on one startup, then this guard would
// otherwise re-enter on the next because no email-tracked steps exist yet.)
try {
const rpCols = db.prepare("PRAGMA table_info(run_profiles)").all() as { name: string }[];
if (!rpCols.some(c => c.name === "state")) return;
} catch { return; }
// Idempotent: skip if run_profile_tracks already has rows or workflow_steps already has email-tracked rows
try {
const alreadyRun = (db.prepare("SELECT COUNT(*) as c FROM run_profile_tracks").get() as { c: number }).c > 0
|| (db.prepare("SELECT COUNT(*) as c FROM workflow_steps WHERE track = 'email'").get() as { c: number }).c > 0;
if (alreadyRun) return;
} catch { return; }
db.transaction(() => {
// 1. Assign email step_type rows to the email track
db.exec("UPDATE workflow_steps SET track = 'email' WHERE step_type = 'email'");
// Also assign delay steps to the email track if the next non-delay step after them is an email step.
// Without this, delays between email steps default to 'linkedin' and create a ghost linkedin track.
db.exec(`
UPDATE workflow_steps SET track = 'email'
WHERE step_type = 'delay'
AND (
SELECT step_type FROM workflow_steps ws2
WHERE ws2.workflow_id = workflow_steps.workflow_id
AND ws2.step_order > workflow_steps.step_order
AND ws2.step_type != 'delay'
ORDER BY ws2.step_order ASC
LIMIT 1
) = 'email'
`);
// 2. Re-number step_order densely within each (workflow_id, track), preserving original order
const stepGroups = db.prepare(
"SELECT id, workflow_id, track, step_order FROM workflow_steps ORDER BY workflow_id, track, step_order"
).all() as Array<{ id: string; workflow_id: string; track: string; step_order: number }>;
// Group by (workflow_id, track) and assign dense 1-based order
const grouped = new Map<string, Array<{ id: string; step_order: number }>>();
for (const row of stepGroups) {
const key = `${row.workflow_id}|${row.track}`;
if (!grouped.has(key)) grouped.set(key, []);
grouped.get(key)!.push({ id: row.id, step_order: row.step_order });
}
const updateStep = db.prepare("UPDATE workflow_steps SET step_order = ? WHERE id = ?");
for (const steps of grouped.values()) {
steps.sort((a, b) => a.step_order - b.step_order);
steps.forEach((s, i) => updateStep.run(i + 1, s.id));
}
// 3. Backfill run_profile_tracks from existing run_profiles
// Only backfill if there are run_profiles rows to process
const allProfiles = db.prepare(
`SELECT rp.id, rp.run_id, rp.state, rp.current_step, rp.next_step_at,
rp.error_message, rp.last_email_subject, rp.last_email_body, rp.last_linkedin_message,
r.workflow_id
FROM run_profiles rp
JOIN runs r ON r.id = rp.run_id`
).all() as Array<{
id: string; run_id: string; state: string; current_step: number;
next_step_at: string | null; error_message: string | null;
last_email_subject: string | null; last_email_body: string | null;
last_linkedin_message: string | null; workflow_id: string;
}>;
if (allProfiles.length === 0) return;
// Load all workflow steps grouped by workflow_id, preserving their original ordering
// We need the ORIGINAL step_order to map legacy current_step (0-based flat index) to tracks.
// After the re-numbering above, step_order is now per-track. We stored the original order as the
// sort key inside stepGroups above. Rebuild a per-workflow flat list from the original ordering.
const workflowStepsOrig = db.prepare(
"SELECT id, workflow_id, track, step_order FROM workflow_steps ORDER BY workflow_id, step_order"
).all() as Array<{ id: string; workflow_id: string; track: string; step_order: number }>;
// For each workflow, build the flat list of steps in their original order (by new step_order within track, then track order linkedin < email)
// BUT: we need the ORIGINAL flat order before re-numbering. Since we already re-numbered, we have to reconstruct.
// Approach: the original flat step_order was cross-track. The re-numbered step_order is per-track and starts at 1.
// We stored the original step_order in stepGroups (before modification). Use that.
const origStepOrderMap = new Map<string, number>(); // step id → original flat step_order
for (const row of stepGroups) {
origStepOrderMap.set(row.id, row.step_order);
}
// Per workflow: flat list of steps sorted by original step_order
const workflowFlatSteps = new Map<string, Array<{ id: string; track: string; orig_order: number }>>();
for (const step of workflowStepsOrig) {
if (!workflowFlatSteps.has(step.workflow_id)) workflowFlatSteps.set(step.workflow_id, []);
workflowFlatSteps.get(step.workflow_id)!.push({
id: step.id,
track: step.track,
orig_order: origStepOrderMap.get(step.id) ?? step.step_order,
});
}
for (const steps of workflowFlatSteps.values()) {
steps.sort((a, b) => a.orig_order - b.orig_order);
}
// Per workflow: which tracks exist
const workflowTracks = new Map<string, Set<string>>();
for (const step of workflowStepsOrig) {
if (!workflowTracks.has(step.workflow_id)) workflowTracks.set(step.workflow_id, new Set());
workflowTracks.get(step.workflow_id)!.add(step.track);
}
const insertTrack = db.prepare(`
INSERT OR IGNORE INTO run_profile_tracks
(id, run_profile_id, track, state, current_step, last_step_at, next_step_at,
error_message, last_email_subject, last_email_body, last_linkedin_message)
VALUES (?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?)
`);
for (const rp of allProfiles) {
const flatSteps = workflowFlatSteps.get(rp.workflow_id) ?? [];
const tracks = workflowTracks.get(rp.workflow_id) ?? new Set(["linkedin"]);
const terminalStates = new Set(["completed", "failed", "skipped"]);
if (terminalStates.has(rp.state)) {
// Terminal profile — mark all tracks with the same terminal state
for (const track of tracks) {
insertTrack.run(
randomUUID(), rp.id, track, rp.state, 0,
null, rp.error_message ?? null, null, null, null
);
}
continue;
}
// Active profile — compute per-track completed step count
// legacy current_step is 0-based index into the flat step list = steps already completed
const completedCount = rp.current_step;
const completedSteps = flatSteps.slice(0, completedCount);
const trackCompletedCount = new Map<string, number>();
for (const track of tracks) trackCompletedCount.set(track, 0);
for (const s of completedSteps) {
trackCompletedCount.set(s.track, (trackCompletedCount.get(s.track) ?? 0) + 1);
}
// Which track owned the step we were waiting on (index = completedCount)?
const currentFlatStep = flatSteps[completedCount];
const waitingTrack = currentFlatStep?.track ?? null;
for (const track of tracks) {
const trackCurrentStep = trackCompletedCount.get(track) ?? 0;
// next_step_at: only carry over to the track that was waiting; other track starts immediately
const nextStepAt = track === waitingTrack ? rp.next_step_at : null;
const lastEmailSubject = track === "email" ? rp.last_email_subject : null;
const lastEmailBody = track === "email" ? rp.last_email_body : null;
const lastLinkedinMessage = track === "linkedin" ? rp.last_linkedin_message : null;
insertTrack.run(
randomUUID(), rp.id, track, rp.state, trackCurrentStep,
nextStepAt, rp.error_message ?? null,
lastEmailSubject, lastEmailBody, lastLinkedinMessage
);
}
}
})();
}
function dropDeprecatedRunProfileColumns(db: Database.Database) {
// Idempotent: check if state column still exists on run_profiles
try {
const tableInfo = db.prepare("PRAGMA table_info(run_profiles)").all() as { name: string }[];
const hasState = tableInfo.some(c => c.name === "state");
if (!hasState) return; // already dropped
} catch { return; }
// SQLite requires a table rebuild to drop columns
try {
db.exec(`
PRAGMA foreign_keys = OFF;
CREATE TABLE run_profiles_new (
id TEXT PRIMARY KEY,
run_id TEXT REFERENCES runs(id) ON DELETE CASCADE,
target_id TEXT REFERENCES targets(id),
email_account_id TEXT,
created_at TEXT DEFAULT (datetime('now')),
UNIQUE(run_id, target_id)
);
INSERT INTO run_profiles_new (id, run_id, target_id, email_account_id, created_at)
SELECT id, run_id, target_id, email_account_id, created_at FROM run_profiles;
DROP TABLE run_profiles;
ALTER TABLE run_profiles_new RENAME TO run_profiles;
PRAGMA foreign_keys = ON;
`);
} catch { /* ignore — may already be done */ }
}
function runMigrations(db: Database.Database) {
// Add columns introduced after initial schema — safe to run on existing DBs
const migrations = [
"ALTER TABLE targets ADD COLUMN degree INTEGER",
"ALTER TABLE targets ADD COLUMN connection_requested_at TEXT",
"ALTER TABLE targets ADD COLUMN connected_at TEXT",
"ALTER TABLE targets ADD COLUMN message_sent_at TEXT",
"ALTER TABLE targets ADD COLUMN last_replied_at TEXT",
"ALTER TABLE targets ADD COLUMN linkedin_member_urn TEXT",
"ALTER TABLE targets ADD COLUMN sales_nav_url TEXT",
"ALTER TABLE lists ADD COLUMN sales_nav_url TEXT",
"ALTER TABLE accounts ADD COLUMN inbox_synced_at TEXT",
"ALTER TABLE accounts ADD COLUMN active_hours_start INTEGER DEFAULT 9",
"ALTER TABLE accounts ADD COLUMN active_hours_end INTEGER DEFAULT 18",
"ALTER TABLE accounts ADD COLUMN timezone TEXT DEFAULT 'UTC'",
"ALTER TABLE accounts ADD COLUMN working_days TEXT DEFAULT '1,2,3,4,5'",
"ALTER TABLE workflow_steps ADD COLUMN connect_note TEXT",
"ALTER TABLE workflow_steps ADD COLUMN message_body TEXT",
"ALTER TABLE targets ADD COLUMN headline TEXT",
"ALTER TABLE targets ADD COLUMN summary TEXT",
"ALTER TABLE accounts ADD COLUMN accepted_sync_at TEXT",
// Messaging identity (fsd_profile URN, "urn:li:fsd_profile:ACoAA...") — the
// form the messaging GraphQL API returns. It is NOT convertible to the
// numeric urn:li:member we store elsewhere, so we capture it on first reply
// (matched by name) and then join by it directly on every later sync.
"ALTER TABLE targets ADD COLUMN messaging_urn TEXT",
"CREATE INDEX IF NOT EXISTS idx_targets_messaging_urn ON targets(messaging_urn)",
// Boundary for the incremental connections sync: the createdAt of the newest
// connection seen last run. NULL = never synced (first run does a full pass).
"ALTER TABLE accounts ADD COLUMN connections_synced_through_ms INTEGER",
"ALTER TABLE accounts ADD COLUMN li_connections INTEGER",
"ALTER TABLE accounts ADD COLUMN li_pending INTEGER",
"ALTER TABLE accounts ADD COLUMN li_profile_views INTEGER",
"ALTER TABLE accounts ADD COLUMN li_stats_synced_at TEXT",
`CREATE TABLE IF NOT EXISTS workflow_step_templates (
step_id TEXT REFERENCES workflow_steps(id) ON DELETE CASCADE,
template_id TEXT REFERENCES templates(id) ON DELETE CASCADE,
PRIMARY KEY (step_id, template_id)
)`,
// Extended lead data from salesApiLeadSearch
"ALTER TABLE targets ADD COLUMN object_urn TEXT",
"ALTER TABLE targets ADD COLUMN open_link INTEGER DEFAULT 0",
"ALTER TABLE targets ADD COLUMN company_industry TEXT",
"ALTER TABLE targets ADD COLUMN company_location TEXT",
"ALTER TABLE targets ADD COLUMN tenure_months INTEGER",
"ALTER TABLE targets ADD COLUMN spotlight_badges TEXT",
// Profile enrichment — populated by visiting their Sales Nav profile page
"ALTER TABLE targets ADD COLUMN positions_json TEXT",
"ALTER TABLE targets ADD COLUMN skills_json TEXT",
"ALTER TABLE targets ADD COLUMN enriched_profile_at TEXT",
// Email outreach fields
"ALTER TABLE targets ADD COLUMN email TEXT",
"ALTER TABLE targets ADD COLUMN email_replied_at TEXT",
"ALTER TABLE targets ADD COLUMN company_id TEXT",
// Email account on runs (nullable — only needed when workflow has email steps)
"ALTER TABLE runs ADD COLUMN email_account_id TEXT REFERENCES email_accounts(id)",
// Workflow step email fields
"ALTER TABLE workflow_steps ADD COLUMN email_subject TEXT",
"ALTER TABLE workflow_steps ADD COLUMN email_body TEXT",
// Apollo enrichment fields
"ALTER TABLE targets ADD COLUMN apollo_id TEXT",
"ALTER TABLE targets ADD COLUMN seniority TEXT",
"ALTER TABLE targets ADD COLUMN apollo_functions TEXT",
"ALTER TABLE targets ADD COLUMN company_description TEXT",
"ALTER TABLE targets ADD COLUMN company_size INTEGER",
"ALTER TABLE targets ADD COLUMN apollo_enriched_at TEXT",
"ALTER TABLE targets ADD COLUMN email_status TEXT",
// Manual fields
"ALTER TABLE targets ADD COLUMN notes TEXT",
// Apollo extra person fields
"ALTER TABLE targets ADD COLUMN city TEXT",
"ALTER TABLE targets ADD COLUMN country TEXT",
"ALTER TABLE targets ADD COLUMN time_zone TEXT",
"ALTER TABLE targets ADD COLUMN apollo_departments TEXT",
// Apollo extra company fields on companies table
"ALTER TABLE companies ADD COLUMN founded_year INTEGER",
"ALTER TABLE companies ADD COLUMN logo_url TEXT",
"ALTER TABLE companies ADD COLUMN phone TEXT",
"ALTER TABLE companies ADD COLUMN annual_revenue TEXT",
"ALTER TABLE companies ADD COLUMN technology_names TEXT",
"ALTER TABLE companies ADD COLUMN keywords TEXT",
"ALTER TABLE companies ADD COLUMN city TEXT",
"ALTER TABLE companies ADD COLUMN country TEXT",
// Email signature
"ALTER TABLE email_accounts ADD COLUMN signature TEXT",
// Reply-To override — if set, outgoing emails include Reply-To header
"ALTER TABLE email_accounts ADD COLUMN reply_to TEXT",
// Ramp-up: start slow and increase sending volume over time
"ALTER TABLE email_accounts ADD COLUMN ramp_up_enabled INTEGER DEFAULT 1",
"ALTER TABLE email_accounts ADD COLUMN ramp_start_date TEXT",
// Company description and employee count moved from targets to companies
"ALTER TABLE companies ADD COLUMN description TEXT",
"ALTER TABLE companies ADD COLUMN employee_count INTEGER",
// AI agent columns on workflow steps
"ALTER TABLE workflow_steps ADD COLUMN ai_enabled INTEGER DEFAULT 0",
"ALTER TABLE workflow_steps ADD COLUMN ai_model TEXT",
"ALTER TABLE workflow_steps ADD COLUMN ai_prompt TEXT",
"ALTER TABLE workflow_steps ADD COLUMN ai_max_words INTEGER",
// Email step position — which followup number (1 = cold email, 2 = first followup, etc.)
"ALTER TABLE workflow_steps ADD COLUMN email_position INTEGER DEFAULT 1",
// Agent default model (stored on agent_config)
"ALTER TABLE agent_config ADD COLUMN default_model TEXT",
// Email threading — store sent email message-id for reply threading (future use)
"ALTER TABLE run_profiles ADD COLUMN last_email_subject TEXT",
"ALTER TABLE run_profiles ADD COLUMN last_email_body TEXT",
// LinkedIn message follow-up tracking
"ALTER TABLE workflow_steps ADD COLUMN message_position INTEGER DEFAULT 1",
"ALTER TABLE run_profiles ADD COLUMN last_linkedin_message TEXT",
// Language for AI-generated content per step
"ALTER TABLE workflow_steps ADD COLUMN ai_language TEXT DEFAULT 'English'",
// Campaign-level prompt — per-workflow AI context (USP, persona, tone for this campaign)
"ALTER TABLE workflows ADD COLUMN prompt TEXT",
// Email domain invalid flag on companies — set when a bounce is detected for any contact at this company
"ALTER TABLE companies ADD COLUMN email_domain_invalid INTEGER DEFAULT 0",
// Apollo extra person fields
"ALTER TABLE targets ADD COLUMN email_domain_catchall INTEGER DEFAULT 0",
// Per-profile email account assignment (multi-account routing)
"ALTER TABLE run_profiles ADD COLUMN email_account_id TEXT",
// Backfill: copy email_account_id from runs → run_profiles for existing records
`UPDATE run_profiles SET email_account_id = (SELECT email_account_id FROM runs WHERE runs.id = run_profiles.run_id) WHERE email_account_id IS NULL`,
// Import job progress tracking
`CREATE TABLE IF NOT EXISTS list_imports (
id TEXT PRIMARY KEY,
list_id TEXT NOT NULL REFERENCES lists(id) ON DELETE CASCADE,
status TEXT NOT NULL DEFAULT 'running',
phase TEXT,
page INTEGER DEFAULT 0,
total_pages INTEGER DEFAULT 0,
count INTEGER DEFAULT 0,
total INTEGER DEFAULT 0,
imported INTEGER DEFAULT 0,
skipped INTEGER DEFAULT 0,
error TEXT,
started_at TEXT NOT NULL DEFAULT (datetime('now')),
finished_at TEXT
)`,
// Parallel tracks: add track column to workflow_steps
"ALTER TABLE workflow_steps ADD COLUMN track TEXT NOT NULL DEFAULT 'linkedin' CHECK(track IN ('linkedin', 'email'))",
// Parallel tracks: create run_profile_tracks table
`CREATE TABLE IF NOT EXISTS run_profile_tracks (
id TEXT PRIMARY KEY,
run_profile_id TEXT NOT NULL REFERENCES run_profiles(id) ON DELETE CASCADE,
track TEXT NOT NULL CHECK(track IN ('linkedin', 'email')),
state TEXT NOT NULL DEFAULT 'pending' CHECK(state IN ('pending', 'in_progress', 'completed', 'failed', 'skipped')),
current_step INTEGER NOT NULL DEFAULT 0,
last_step_at TEXT,
next_step_at TEXT,
error_message TEXT,
last_email_subject TEXT,
last_email_body TEXT,
last_linkedin_message TEXT,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
UNIQUE(run_profile_id, track)
)`,
"CREATE INDEX IF NOT EXISTS idx_run_profile_tracks_run_profile_id ON run_profile_tracks(run_profile_id)",
"CREATE INDEX IF NOT EXISTS idx_run_profile_tracks_state_next ON run_profile_tracks(state, next_step_at)",
// Drop deprecated columns from run_profiles — all consumers now read from run_profile_tracks
// SQLite does not support DROP COLUMN directly before 3.35; handled via table rebuild below
// Separate IMAP credentials for custom mail providers where SMTP ≠ IMAP auth
"ALTER TABLE email_accounts ADD COLUMN imap_username TEXT",
"ALTER TABLE email_accounts ADD COLUMN imap_password TEXT",
"ALTER TABLE workflows ADD COLUMN is_archived INTEGER NOT NULL DEFAULT 0",
// Per-step signature override for email steps (null = use email account default)
"ALTER TABLE workflow_steps ADD COLUMN email_signature TEXT",
// CRM: todos per contact
`CREATE TABLE IF NOT EXISTS todos (
id TEXT PRIMARY KEY,
target_id TEXT NOT NULL REFERENCES targets(id) ON DELETE CASCADE,
title TEXT NOT NULL,
due_date TEXT,
status TEXT NOT NULL DEFAULT 'open' CHECK(status IN ('open', 'done')),
created_at TEXT NOT NULL DEFAULT (datetime('now'))
)`,
"CREATE INDEX IF NOT EXISTS idx_todos_target_id ON todos(target_id)",
"CREATE INDEX IF NOT EXISTS idx_todos_status ON todos(status)",
"ALTER TABLE todos ADD COLUMN description TEXT",
// CRM: activity log per contact
`CREATE TABLE IF NOT EXISTS activity_logs (
id TEXT PRIMARY KEY,
target_id TEXT NOT NULL REFERENCES targets(id) ON DELETE CASCADE,
type TEXT NOT NULL DEFAULT 'note' CHECK(type IN ('call', 'email', 'meeting', 'note', 'other')),
body TEXT NOT NULL,
logged_at TEXT NOT NULL DEFAULT (datetime('now')),
created_at TEXT NOT NULL DEFAULT (datetime('now'))
)`,
"CREATE INDEX IF NOT EXISTS idx_activity_logs_target_id ON activity_logs(target_id)",
// Reply classifier: captured email replies + classifier verdict + dispatcher result
`CREATE TABLE IF NOT EXISTS email_replies (
id TEXT PRIMARY KEY,
target_id TEXT NOT NULL REFERENCES targets(id) ON DELETE CASCADE,
run_id TEXT REFERENCES runs(id) ON DELETE SET NULL,
from_email TEXT NOT NULL,
subject TEXT,
body_text TEXT NOT NULL,
received_at TEXT NOT NULL,
classified_at TEXT,
classification_json TEXT,
classification_error TEXT,
dispatched_at TEXT,
dispatch_result_json TEXT,
manually_edited INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
)`,
"CREATE INDEX IF NOT EXISTS idx_email_replies_target_id ON email_replies(target_id)",
"CREATE INDEX IF NOT EXISTS idx_email_replies_dispatched_at ON email_replies(dispatched_at)",
// Reply classifier verdict stamped on the target — enables reply-rate-by-kind metrics
"ALTER TABLE targets ADD COLUMN reply_kind TEXT",
"ALTER TABLE targets ADD COLUMN inmail_sent_at TEXT",
"ALTER TABLE targets ADD COLUMN posts_json TEXT", // recent posts from visit_profile
"ALTER TABLE targets ADD COLUMN posts_scraped_at TEXT",
// One-shot OOO reply context for the AI follow-up writer — set by the dispatcher,
// read + cleared by the runner on the next email send. Distinct from last_email_body
// (which holds the last email WE sent, used for follow-up threading).
"ALTER TABLE run_profile_tracks ADD COLUMN pending_reply_context TEXT",
// Removed the in-app chat agent (replaced by the hosted MCP endpoint at /api/mcp) — drop its tables.
"DROP TABLE IF EXISTS chat_messages",
"DROP TABLE IF EXISTS chat_sessions",
// Batched/scheduled imports — split large lists across days under a daily cap.
"ALTER TABLE list_imports ADD COLUMN account_id TEXT",
"ALTER TABLE list_imports ADD COLUMN sales_nav_url TEXT",
"ALTER TABLE list_imports ADD COLUMN scheduled_for TEXT", // 'YYYY-MM-DD'; NULL = run now
"ALTER TABLE list_imports ADD COLUMN start_page INTEGER DEFAULT 1",
"ALTER TABLE list_imports ADD COLUMN cap INTEGER", // max contacts for this batch
"ALTER TABLE list_imports ADD COLUMN cancel_requested INTEGER DEFAULT 0",
"ALTER TABLE list_imports ADD COLUMN batch_index INTEGER DEFAULT 1",
"ALTER TABLE list_imports ADD COLUMN enrich INTEGER DEFAULT 0",
"CREATE INDEX IF NOT EXISTS idx_list_imports_scheduled ON list_imports(status, scheduled_for)",
// Simple app-wide key/value settings (e.g. daily_import_cap)
`CREATE TABLE IF NOT EXISTS app_settings (
key TEXT PRIMARY KEY,
value TEXT,
updated_at TEXT NOT NULL DEFAULT (datetime('now'))
)`,
// Sales Nav InMail gets its own daily budget — separate from daily_message_limit,
// since InMail (non-connections) and regular messages (connections) were being
// gated off the same counter, starving one whenever the other was busy.
"ALTER TABLE accounts ADD COLUMN daily_inmail_limit INTEGER DEFAULT 15",
// Sales Nav search: persistent cache of resolved filter values (typeahead
// ids). Dedup on (filter_type, id) — LinkedIn ids are stable (e.g. Berlin's
// geoUrn never changes), so once resolved a value is reused forever with no
// further live typeahead calls. Lives here (not ee/) per the open-core rule:
// all migrations stay in lib/db.ts. Consumed only by the ee/ search feature.
`CREATE TABLE IF NOT EXISTS search_filter_cache (
id TEXT NOT NULL,
filter_type TEXT NOT NULL,
display_value TEXT NOT NULL,
headline TEXT,
query TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
PRIMARY KEY (filter_type, id)
)`,
"CREATE INDEX IF NOT EXISTS idx_sfc_type_query ON search_filter_cache(filter_type, query)",
"CREATE INDEX IF NOT EXISTS idx_sfc_type_display ON search_filter_cache(filter_type, display_value)",
// CSV import: lists can be flagged as linkedin- or email-only, so the runner/UI
// can warn before enrolling a purpose-mismatched list into a campaign.
"ALTER TABLE lists ADD COLUMN purpose TEXT",
// Manual/CSV-only field — no automation reads or writes this, reference data only.
"ALTER TABLE targets ADD COLUMN phone TEXT",
];
for (const sql of migrations) {
try { db.exec(sql); } catch { /* column already exists */ }
}
// Parallel tracks: assign email steps to email track, re-number step_order, backfill run_profile_tracks
runParallelTracksMigration(db);
// Drop deprecated run_profiles columns (state, current_step, etc.) — consumers now read track-runs
dropDeprecatedRunProfileColumns(db);
// Migrate workflow_steps CHECK constraint to allow 'delay' and 'email' step_types
try {
const tableInfo = db.prepare("SELECT sql FROM sqlite_master WHERE type='table' AND name='workflow_steps'").get() as { sql: string } | undefined;
if (tableInfo && (!tableInfo.sql.includes("'delay'") || !tableInfo.sql.includes("'email'"))) {
db.exec(`
PRAGMA foreign_keys = OFF;
CREATE TABLE workflow_steps_new (
id TEXT PRIMARY KEY,
workflow_id TEXT REFERENCES workflows(id) ON DELETE CASCADE,
step_order INTEGER NOT NULL,
step_type TEXT NOT NULL CHECK(step_type IN ('visit', 'connect', 'message', 'delay', 'email')),
template_id TEXT REFERENCES templates(id),
delay_seconds INTEGER DEFAULT 0,
connect_note TEXT,
message_body TEXT,
email_subject TEXT,
email_body TEXT,
enabled INTEGER DEFAULT 1
);
INSERT INTO workflow_steps_new
SELECT id, workflow_id, step_order, step_type, template_id, delay_seconds,
connect_note, message_body,
NULL, NULL,
enabled
FROM workflow_steps;
DROP TABLE workflow_steps;
ALTER TABLE workflow_steps_new RENAME TO workflow_steps;
PRAGMA foreign_keys = ON;
`);
}
} catch { /* migration already done */ }
// Allow the 'sales_inmail' step_type (Sales Navigator InMail). Rebuilds the
// table preserving EVERY current column (the historical rebuild above only
// copied the original columns — do NOT reuse it). InMail reuses message_body
// for the body and email_subject for the required subject.
try {
const ti = db.prepare("SELECT sql FROM sqlite_master WHERE type='table' AND name='workflow_steps'").get() as { sql: string } | undefined;
if (ti && !ti.sql.includes("'sales_inmail'")) {
const cols = (db.prepare("PRAGMA table_info(workflow_steps)").all() as Array<{ name: string }>).map((c) => c.name);
const colList = cols.join(", ");
db.exec(`
PRAGMA foreign_keys = OFF;
CREATE TABLE workflow_steps_new (
id TEXT PRIMARY KEY,
workflow_id TEXT REFERENCES workflows(id) ON DELETE CASCADE,
step_order INTEGER NOT NULL,
step_type TEXT NOT NULL CHECK(step_type IN ('visit', 'connect', 'message', 'sales_inmail', 'delay', 'email')),
template_id TEXT REFERENCES templates(id),
delay_seconds INTEGER DEFAULT 0,
connect_note TEXT,
message_body TEXT,
email_subject TEXT,
email_body TEXT,
enabled INTEGER DEFAULT 1,
ai_enabled INTEGER DEFAULT 0,
ai_model TEXT,
ai_prompt TEXT,
ai_max_words INTEGER,
email_position INTEGER DEFAULT 1,
message_position INTEGER DEFAULT 1,
ai_language TEXT DEFAULT 'English',
track TEXT NOT NULL DEFAULT 'linkedin' CHECK(track IN ('linkedin', 'email')),
email_signature TEXT
);
INSERT INTO workflow_steps_new (${colList}) SELECT ${colList} FROM workflow_steps;
DROP TABLE workflow_steps;
ALTER TABLE workflow_steps_new RENAME TO workflow_steps;
PRAGMA foreign_keys = ON;
`);
}
} catch { /* migration already done */ }
// CSV import: allow email-only targets (no LinkedIn URL). targets.linkedin_url was
// NOT NULL UNIQUE from the base schema — rebuild to make it nullable (still UNIQUE,
// SQLite allows multiple NULLs under UNIQUE) preserving EVERY current column, same
// pattern as the sales_inmail rebuild above (do not reuse the older historical rebuilds
// that only copied the original columns).
try {
const ti = db.prepare("SELECT sql FROM sqlite_master WHERE type='table' AND name='targets'").get() as { sql: string } | undefined;
if (ti && ti.sql.includes("linkedin_url TEXT NOT NULL")) {
const cols = (db.prepare("PRAGMA table_info(targets)").all() as Array<{ name: string; type: string; notnull: number; dflt_value: string | null }>);
const colDefs = cols.map((c) => {
if (c.name === "linkedin_url") return "linkedin_url TEXT UNIQUE";
if (c.name === "id") return "id TEXT PRIMARY KEY";
const notnull = c.notnull ? " NOT NULL" : "";
// Non-literal defaults (e.g. datetime('now')) must be parenthesized in SQLite DDL.
const isLiteral = c.dflt_value === null || /^-?\d+(\.\d+)?$/.test(c.dflt_value) || /^'.*'$/.test(c.dflt_value);
const dflt = c.dflt_value !== null ? ` DEFAULT ${isLiteral ? c.dflt_value : `(${c.dflt_value})`}` : "";
return `${c.name} ${c.type}${notnull}${dflt}`;
});
const colList = cols.map((c) => c.name).join(", ");
db.exec(`
PRAGMA foreign_keys = OFF;
CREATE TABLE targets_new (
${colDefs.join(",\n ")}
);
INSERT INTO targets_new (${colList}) SELECT ${colList} FROM targets;
DROP TABLE targets;
ALTER TABLE targets_new RENAME TO targets;
PRAGMA foreign_keys = ON;
`);
}
} catch { /* migration already done */ }
// Backfill: move company_description and company_size from targets into companies
try {
db.exec(`
UPDATE companies
SET
description = COALESCE(description, (
SELECT t.company_description FROM targets t
WHERE t.company_id = companies.id AND t.company_description IS NOT NULL
LIMIT 1
)),
employee_count = COALESCE(employee_count, (
SELECT t.company_size FROM targets t
WHERE t.company_id = companies.id AND t.company_size IS NOT NULL
LIMIT 1
))
WHERE description IS NULL OR employee_count IS NULL
`);
} catch { /* ignore */ }
// Backfill: for old records where linkedin_url is a Sales Nav URL, move it to sales_nav_url
try {
db.exec(`
UPDATE targets
SET sales_nav_url = linkedin_url
WHERE linkedin_url LIKE '%/sales/lead/%' AND (sales_nav_url IS NULL OR sales_nav_url = '')
`);
} catch { /* ignore */ }
// Create unique index on run_profiles if not already present (idempotent)
try {
db.exec("CREATE UNIQUE INDEX IF NOT EXISTS idx_run_profiles_unique ON run_profiles(run_id, target_id);");
} catch { /* ignore */ }
encryptLegacySecretsMigration(db);
}
// One-time (per-row) migration: encrypt any plaintext accounts.cookies_json,
// email_accounts.password/imap_password, integrations.api_key left over from before
// encryption-at-rest was added. isEncrypted() lets this run safely on every boot — already
//-encrypted rows (real "v1:" ciphertext) are skipped, so this is idempotent and cheap once
// migrated. See lib/crypto.ts for the format and lib/premium.ts-style boundary notes.
function encryptLegacySecretsMigration(db: Database.Database) {
const accounts = db.prepare("SELECT id, cookies_json FROM accounts WHERE cookies_json IS NOT NULL").all() as
{ id: string; cookies_json: string }[];
for (const row of accounts) {
if (isEncrypted(row.cookies_json)) continue;
db.prepare("UPDATE accounts SET cookies_json = ? WHERE id = ?").run(encryptSecret(row.cookies_json), row.id);
}
const emailAccounts = db
.prepare("SELECT id, password, imap_password FROM email_accounts")
.all() as { id: string; password: string; imap_password: string | null }[];
for (const row of emailAccounts) {
const needsPassword = !isEncrypted(row.password);
const needsImapPassword = row.imap_password !== null && !isEncrypted(row.imap_password);
if (!needsPassword && !needsImapPassword) continue;
db.prepare("UPDATE email_accounts SET password = ?, imap_password = ? WHERE id = ?").run(
needsPassword ? encryptSecret(row.password) : row.password,
needsImapPassword ? encryptSecret(row.imap_password!) : row.imap_password,
row.id
);
}
const integrations = db.prepare("SELECT key, api_key FROM integrations WHERE api_key IS NOT NULL").all() as
{ key: string; api_key: string }[];
for (const row of integrations) {
if (isEncrypted(row.api_key)) continue;
db.prepare("UPDATE integrations SET api_key = ? WHERE key = ?").run(encryptSecret(row.api_key), row.key);
}
}
function initDb(db: Database.Database) {
db.exec(`
CREATE TABLE IF NOT EXISTS accounts (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
email TEXT NOT NULL UNIQUE,
cookies_json TEXT,
is_authenticated INTEGER DEFAULT 0,
daily_connection_limit INTEGER DEFAULT 20,
daily_message_limit INTEGER DEFAULT 50,
daily_inmail_limit INTEGER DEFAULT 15,
active_hours_start INTEGER DEFAULT 9,
active_hours_end INTEGER DEFAULT 18,
timezone TEXT DEFAULT 'UTC',
working_days TEXT DEFAULT '1,2,3,4,5',
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS targets (
id TEXT PRIMARY KEY,
linkedin_url TEXT UNIQUE,
sales_nav_url TEXT,
first_name TEXT,
last_name TEXT,
full_name TEXT,
title TEXT,
company TEXT,
location TEXT,
profile_image_url TEXT,
degree INTEGER,
connection_requested_at TEXT,
connected_at TEXT,
message_sent_at TEXT,
last_replied_at TEXT,
linkedin_member_urn TEXT,
enriched_at TEXT,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS lists (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
description TEXT,
sales_nav_url TEXT,
purpose TEXT CHECK(purpose IN ('linkedin', 'email')),
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS list_targets (
list_id TEXT REFERENCES lists(id) ON DELETE CASCADE,
target_id TEXT REFERENCES targets(id) ON DELETE CASCADE,
PRIMARY KEY (list_id, target_id)
);
CREATE TABLE IF NOT EXISTS templates (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
body TEXT NOT NULL,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS workflows (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
description TEXT,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS workflow_steps (
id TEXT PRIMARY KEY,
workflow_id TEXT REFERENCES workflows(id) ON DELETE CASCADE,
step_order INTEGER NOT NULL,
step_type TEXT NOT NULL CHECK(step_type IN ('visit', 'connect', 'message', 'delay')),
template_id TEXT REFERENCES templates(id),
delay_seconds INTEGER DEFAULT 0,
connect_note TEXT,
message_body TEXT,
enabled INTEGER DEFAULT 1
);
CREATE TABLE IF NOT EXISTS workflow_step_templates (
step_id TEXT REFERENCES workflow_steps(id) ON DELETE CASCADE,
template_id TEXT REFERENCES templates(id) ON DELETE CASCADE,
PRIMARY KEY (step_id, template_id)
);
CREATE TABLE IF NOT EXISTS runs (
id TEXT PRIMARY KEY,
workflow_id TEXT REFERENCES workflows(id),
list_id TEXT REFERENCES lists(id),
account_id TEXT REFERENCES accounts(id),
status TEXT DEFAULT 'pending' CHECK(status IN ('pending', 'running', 'paused', 'completed', 'failed')),
created_at TEXT DEFAULT (datetime('now')),
started_at TEXT,
completed_at TEXT,
runner_pid INTEGER
);
CREATE TABLE IF NOT EXISTS run_profiles (
id TEXT PRIMARY KEY,
run_id TEXT REFERENCES runs(id) ON DELETE CASCADE,
target_id TEXT REFERENCES targets(id),
state TEXT DEFAULT 'pending' CHECK(state IN ('pending', 'in_progress', 'completed', 'failed', 'skipped')),
current_step INTEGER DEFAULT 0,
last_step_at TEXT,
next_step_at TEXT,
error_message TEXT,
created_at TEXT DEFAULT (datetime('now')),
UNIQUE(run_id, target_id)
);
CREATE TABLE IF NOT EXISTS logs (
id TEXT PRIMARY KEY,
run_id TEXT REFERENCES runs(id) ON DELETE CASCADE,
target_id TEXT REFERENCES targets(id),
level TEXT DEFAULT 'info' CHECK(level IN ('info', 'warn', 'error')),
message TEXT NOT NULL,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS users (
id TEXT PRIMARY KEY,
email TEXT NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS companies (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
domain TEXT,
industry TEXT,
location TEXT,
linkedin_url TEXT,
website TEXT,
notes TEXT,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS integrations (
key TEXT PRIMARY KEY,
api_key TEXT,
created_at TEXT DEFAULT (datetime('now')),
updated_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS agent_config (
id INTEGER PRIMARY KEY DEFAULT 1,
system_prompt TEXT,
user_prompt TEXT,
email_examples TEXT,
linkedin_examples TEXT,
updated_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS agent_sessions (
id TEXT PRIMARY KEY,
run_id TEXT,
target_id TEXT,
step_id TEXT,
model TEXT,
input_tokens INTEGER,
output_tokens INTEGER,
cost_usd REAL,
prompt TEXT,
generated_text TEXT,
created_at TEXT DEFAULT (datetime('now'))
);
-- Hosted MCP OAuth 2.1 server: clients (DCR), single-use auth codes, access/refresh tokens.
-- Tokens/codes are stored only as sha256 hashes. Reuses NEXTAUTH user identity.
CREATE TABLE IF NOT EXISTS oauth_clients (
client_id TEXT PRIMARY KEY,
client_name TEXT,
redirect_uris TEXT NOT NULL,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS oauth_auth_codes (
code_hash TEXT PRIMARY KEY,
client_id TEXT NOT NULL,
user_id TEXT NOT NULL,
redirect_uri TEXT NOT NULL,
code_challenge TEXT NOT NULL,
scope TEXT,
expires_at TEXT NOT NULL,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS oauth_tokens (
id TEXT PRIMARY KEY,
access_hash TEXT NOT NULL UNIQUE,
refresh_hash TEXT UNIQUE,
client_id TEXT NOT NULL,
user_id TEXT NOT NULL,
scope TEXT,
expires_at TEXT NOT NULL,
created_at TEXT DEFAULT (datetime('now'))
);
CREATE TABLE IF NOT EXISTS email_accounts (
id TEXT PRIMARY KEY,
name TEXT NOT NULL,
from_email TEXT NOT NULL,
from_name TEXT,
smtp_host TEXT NOT NULL,
smtp_port INTEGER DEFAULT 587,
smtp_secure INTEGER DEFAULT 0,
imap_host TEXT,
imap_port INTEGER DEFAULT 993,
username TEXT NOT NULL,
password TEXT NOT NULL,
daily_email_limit INTEGER DEFAULT 50,
active_hours_start INTEGER DEFAULT 9,
active_hours_end INTEGER DEFAULT 18,
timezone TEXT DEFAULT 'UTC',
working_days TEXT DEFAULT '1,2,3,4,5',
is_verified INTEGER DEFAULT 0,
inbox_synced_at TEXT,
created_at TEXT DEFAULT (datetime('now'))
);
`);
}