From e73ee04f79ffb30531d8d0cf83a564d19b89e089 Mon Sep 17 00:00:00 2001 From: Brian O'Kelley Date: Wed, 4 Feb 2026 11:56:53 -0500 Subject: [PATCH] refactor(jobs): Migrate to declarative job scheduler Replaces individual start/stop methods with a declarative configuration pattern for all 12 scheduled background jobs. This reduces boilerplate and makes adding new jobs trivial. Changes: - scheduler.ts: Rewritten with JobConfig interface, register/start/stop methods, and built-in business hours constraint handling - job-definitions.ts: New file with all job configurations including the composite content-curator runner - http.ts: Cleaned up by removing 5 inline job methods (~250 lines) and 8 instance variables for job tracking Closes #716 Co-Authored-By: Claude Opus 4.5 --- .changeset/green-rice-relate.md | 2 + server/src/addie/jobs/job-definitions.ts | 222 +++++++++++ server/src/addie/jobs/scheduler.ts | 469 +++++++++-------------- server/src/http.ts | 372 ++---------------- 4 files changed, 431 insertions(+), 634 deletions(-) create mode 100644 .changeset/green-rice-relate.md create mode 100644 server/src/addie/jobs/job-definitions.ts diff --git a/.changeset/green-rice-relate.md b/.changeset/green-rice-relate.md new file mode 100644 index 0000000000..a845151cc8 --- /dev/null +++ b/.changeset/green-rice-relate.md @@ -0,0 +1,2 @@ +--- +--- diff --git a/server/src/addie/jobs/job-definitions.ts b/server/src/addie/jobs/job-definitions.ts new file mode 100644 index 0000000000..66aae39d32 --- /dev/null +++ b/server/src/addie/jobs/job-definitions.ts @@ -0,0 +1,222 @@ +/** + * Job Definitions + * + * Declarative configuration for all scheduled background jobs. + * Call registerAllJobs() on startup to register them with the scheduler. + */ + +import { jobScheduler } from './scheduler.js'; +import { runDocumentIndexerJob } from './committee-document-indexer.js'; +import { runSummaryGeneratorJob } from './committee-summary-generator.js'; +import { runOutreachScheduler } from '../services/proactive-outreach.js'; +import { enrichMissingOrganizations } from '../../services/enrichment.js'; +import { runMoltbookPosterJob } from './moltbook-poster.js'; +import { runMoltbookEngagementJob } from './moltbook-engagement.js'; +import { runTaskReminderJob } from './task-reminder.js'; +import { runEngagementScoringJob } from './engagement-scoring.js'; +import { runGoalFollowUpJob } from './goal-follow-up.js'; +import { + processPendingResources, + processRssPerspectives, + processCommunityArticles, +} from '../services/content-curator.js'; +import { sendCommunityReplies } from '../services/community-articles.js'; +import { processFeedsToFetch } from '../services/feed-fetcher.js'; +import { processAlerts } from '../services/industry-alerts.js'; +import { sendChannelMessage } from '../../slack/client.js'; + +/** + * Composite runner for content curator that runs multiple sub-tasks sequentially. + * Processes: pending resources, RSS perspectives, community articles, community replies. + * Each sub-task is wrapped in try/catch to allow partial success. + */ +async function runContentCuratorJob() { + const results = { + pendingResources: { processed: 0, succeeded: 0, failed: 0 }, + rssPerspectives: { processed: 0, succeeded: 0, failed: 0 }, + communityArticles: { processed: 0, succeeded: 0, failed: 0 }, + communityReplies: { sent: 0, failed: 0 }, + }; + + try { + results.pendingResources = await processPendingResources({ limit: 5 }); + } catch { + // Error logged by scheduler's top-level handler + } + + try { + results.rssPerspectives = await processRssPerspectives({ limit: 5 }); + } catch { + // Error logged by scheduler's top-level handler + } + + try { + results.communityArticles = await processCommunityArticles({ limit: 5 }); + } catch { + // Error logged by scheduler's top-level handler + } + + try { + results.communityReplies = await sendCommunityReplies(async (channelId, threadTs, text) => { + const result = await sendChannelMessage(channelId, { text, thread_ts: threadTs }); + return result.ok; + }); + } catch { + // Error logged by scheduler's top-level handler + } + + return results; +} + +/** + * Register all job configurations with the scheduler. + * Call this on startup before starting jobs. + */ +export function registerAllJobs(): void { + // Document indexer - indexes Google Docs tracked by committees + jobScheduler.register({ + name: 'document-indexer', + description: 'Document indexer', + interval: { value: 60, unit: 'minutes' }, + initialDelay: { value: 1, unit: 'minutes' }, + runner: runDocumentIndexerJob, + options: { batchSize: 20 }, + shouldLogResult: (r) => r.documentsChecked > 0, + }); + + // Summary generator - generates AI summaries for committees + jobScheduler.register({ + name: 'summary-generator', + description: 'Summary generator', + interval: { value: 24, unit: 'hours' }, + initialDelay: { value: 5, unit: 'minutes' }, + runner: runSummaryGeneratorJob, + options: { batchSize: 10 }, + shouldLogResult: (r) => r.summariesGenerated > 0, + }); + + // Proactive outreach - sends DMs to eligible users (has internal per-user business hours check) + jobScheduler.register({ + name: 'proactive-outreach', + description: 'Proactive outreach', + interval: { value: 30, unit: 'minutes' }, + initialDelay: { value: 2, unit: 'minutes' }, + runner: runOutreachScheduler, + options: { limit: 5 }, + }); + + // Account enrichment - enriches accounts via Lusha API + jobScheduler.register({ + name: 'account-enrichment', + description: 'Account enrichment', + interval: { value: 6, unit: 'hours' }, + initialDelay: { value: 3, unit: 'minutes' }, + runner: enrichMissingOrganizations, + options: { limit: 50, includeEmptyProspects: true }, + shouldLogResult: (r) => r.enriched > 0 || r.failed > 0, + }); + + // Moltbook poster - posts articles to Moltbook + jobScheduler.register({ + name: 'moltbook-poster', + description: 'Moltbook poster', + interval: { value: 2, unit: 'hours' }, + initialDelay: { value: 10, unit: 'minutes' }, + runner: runMoltbookPosterJob, + options: { limit: 1 }, + shouldLogResult: (r) => r.postsCreated > 0, + }); + + // Moltbook engagement - engages with Moltbook discussions + jobScheduler.register({ + name: 'moltbook-engagement', + description: 'Moltbook engagement', + interval: { value: 1, unit: 'hours' }, + initialDelay: { value: 10, unit: 'minutes' }, + runner: runMoltbookEngagementJob, + options: { limit: 3 }, + shouldLogResult: (r) => r.commentsCreated > 0 || r.interestingThreads > 0, + }); + + // Content curator - processes external content for knowledge base + jobScheduler.register({ + name: 'content-curator', + description: 'Content curator', + interval: { value: 5, unit: 'minutes' }, + initialDelay: { value: 30, unit: 'seconds' }, + runner: runContentCuratorJob, + shouldLogResult: (r) => + r.pendingResources.processed > 0 || + r.rssPerspectives.processed > 0 || + r.communityArticles.processed > 0 || + r.communityReplies.sent > 0, + }); + + // Feed fetcher - fetches RSS feeds + jobScheduler.register({ + name: 'feed-fetcher', + description: 'Feed fetcher', + interval: { value: 30, unit: 'minutes' }, + initialDelay: { value: 1, unit: 'minutes' }, + runner: processFeedsToFetch, + shouldLogResult: (r) => r.feedsProcessed > 0, + }); + + // Alert processor - sends industry alerts + jobScheduler.register({ + name: 'alert-processor', + description: 'Alert processor', + interval: { value: 5, unit: 'minutes' }, + initialDelay: { value: 2, unit: 'minutes' }, + runner: processAlerts, + shouldLogResult: (r) => r.alerted > 0, + }); + + // Task reminder - sends task reminders during morning hours + jobScheduler.register({ + name: 'task-reminder', + description: 'Task reminder', + interval: { value: 1, unit: 'hours' }, + runner: runTaskReminderJob, + businessHours: { startHour: 8, endHour: 11, skipWeekends: true }, + shouldLogResult: (r) => r.remindersSent > 0, + }); + + // Engagement scoring - updates engagement scores + jobScheduler.register({ + name: 'engagement-scoring', + description: 'Engagement scoring', + interval: { value: 1, unit: 'hours' }, + initialDelay: { value: 10, unit: 'seconds' }, + runner: runEngagementScoringJob, + }); + + // Goal follow-up - sends follow-up messages during business hours + jobScheduler.register({ + name: 'goal-follow-up', + description: 'Goal follow-up', + interval: { value: 4, unit: 'hours' }, + initialDelay: { value: 3, unit: 'minutes' }, + runner: runGoalFollowUpJob, + businessHours: { startHour: 9, endHour: 18, skipWeekends: true }, + shouldLogResult: (r) => r.followUpsSent > 0 || r.goalsReconciled > 0, + }); +} + +/** + * Job names for conditional startup (e.g., Moltbook jobs only if API key is set) + */ +export const JOB_NAMES = { + DOCUMENT_INDEXER: 'document-indexer', + SUMMARY_GENERATOR: 'summary-generator', + PROACTIVE_OUTREACH: 'proactive-outreach', + ACCOUNT_ENRICHMENT: 'account-enrichment', + MOLTBOOK_POSTER: 'moltbook-poster', + MOLTBOOK_ENGAGEMENT: 'moltbook-engagement', + CONTENT_CURATOR: 'content-curator', + FEED_FETCHER: 'feed-fetcher', + ALERT_PROCESSOR: 'alert-processor', + TASK_REMINDER: 'task-reminder', + ENGAGEMENT_SCORING: 'engagement-scoring', + GOAL_FOLLOW_UP: 'goal-follow-up', +} as const; diff --git a/server/src/addie/jobs/scheduler.ts b/server/src/addie/jobs/scheduler.ts index ca1e10f347..65d9260451 100644 --- a/server/src/addie/jobs/scheduler.ts +++ b/server/src/addie/jobs/scheduler.ts @@ -1,362 +1,263 @@ /** * Job Scheduler * - * Centralized management of scheduled background jobs. - * This module handles starting, stopping, and tracking all periodic jobs. + * Declarative management of scheduled background jobs. + * Jobs are defined as configuration objects and managed centrally. */ import { logger as baseLogger } from '../../logger.js'; -import { runDocumentIndexerJob } from './committee-document-indexer.js'; -import { runSummaryGeneratorJob } from './committee-summary-generator.js'; -import { runOutreachScheduler } from '../services/proactive-outreach.js'; -import { enrichMissingOrganizations } from '../../services/enrichment.js'; -import { runMoltbookPosterJob } from './moltbook-poster.js'; -import { runMoltbookEngagementJob } from './moltbook-engagement.js'; const logger = baseLogger.child({ module: 'job-scheduler' }); -interface ScheduledJob { - name: string; - intervalId: NodeJS.Timeout | null; - initialTimeoutId: NodeJS.Timeout | null; +/** + * Time interval configuration + */ +export interface TimeInterval { + value: number; + unit: 'seconds' | 'minutes' | 'hours'; } /** - * Job Scheduler singleton + * Business hours constraint - job only runs during these hours (ET timezone) */ -class JobScheduler { - private jobs: Map = new Map(); - - /** - * Start the committee document indexer job - * Indexes Google Docs tracked by committees, detects changes, generates summaries - */ - startDocumentIndexer(): void { - const JOB_NAME = 'document-indexer'; - const INTERVAL_MINUTES = 60; // Check every hour - const INITIAL_DELAY_MS = 60000; // 1 minute delay on startup +export interface BusinessHoursConstraint { + /** Hour to start running (0-23) */ + startHour: number; + /** Hour to stop running (0-23, exclusive) */ + endHour: number; + /** Skip Saturdays and Sundays. Default: true */ + skipWeekends?: boolean; +} - const job: ScheduledJob = { - name: JOB_NAME, - intervalId: null, - initialTimeoutId: null, - }; +/** + * Job configuration - declarative definition of a scheduled job + */ +export interface JobConfig, TResult = unknown> { + /** Unique job identifier (kebab-case) */ + name: string; - // Run after a delay on startup - job.initialTimeoutId = setTimeout(async () => { - try { - const result = await runDocumentIndexerJob({ batchSize: 20 }); - if (result.documentsChecked > 0) { - logger.info(result, 'Document indexer: initial run completed'); - } - } catch (err) { - logger.error({ err }, 'Document indexer: initial run failed'); - } - }, INITIAL_DELAY_MS); + /** Human-readable description for logging */ + description: string; - // Then run periodically - job.intervalId = setInterval(async () => { - try { - const result = await runDocumentIndexerJob({ batchSize: 20 }); - if (result.documentsChecked > 0) { - logger.info(result, 'Document indexer: job completed'); - } - } catch (err) { - logger.error({ err }, 'Document indexer: job failed'); - } - }, INTERVAL_MINUTES * 60 * 1000); + /** How often to run the job */ + interval: TimeInterval; - this.jobs.set(JOB_NAME, job); - logger.debug({ intervalMinutes: INTERVAL_MINUTES }, 'Document indexer job started'); - } + /** Delay before first run after startup */ + initialDelay?: TimeInterval; - /** - * Start the proactive outreach job - * Sends DMs to eligible users during business hours - */ - startOutreach(): void { - const JOB_NAME = 'proactive-outreach'; - const INTERVAL_MINUTES = 30; // Check every 30 minutes - const INITIAL_DELAY_MS = 120000; // 2 minute delay on startup + /** The async function to execute */ + runner: (options: TOptions) => Promise; - const job: ScheduledJob = { - name: JOB_NAME, - intervalId: null, - initialTimeoutId: null, - }; + /** Options to pass to the runner */ + options?: TOptions; - // Run after a delay on startup - job.initialTimeoutId = setTimeout(async () => { - try { - await runOutreachScheduler({ limit: 5 }); - } catch (err) { - logger.error({ err }, 'Proactive outreach: initial run failed'); - } - }, INITIAL_DELAY_MS); - - // Then run periodically - job.intervalId = setInterval(async () => { - try { - await runOutreachScheduler({ limit: 5 }); - } catch (err) { - logger.error({ err }, 'Proactive outreach: job failed'); - } - }, INTERVAL_MINUTES * 60 * 1000); - - this.jobs.set(JOB_NAME, job); - logger.debug({ intervalMinutes: INTERVAL_MINUTES }, 'Proactive outreach job started'); - } + /** Only run during these business hours (ET timezone) */ + businessHours?: BusinessHoursConstraint; /** - * Start the committee summary generator job - * Generates AI-powered activity summaries for committees + * Return true to log result at info level, false for debug level. + * If not provided, always logs at debug level. */ - startSummaryGenerator(): void { - const JOB_NAME = 'summary-generator'; - const INTERVAL_HOURS = 24; // Once per day - const INITIAL_DELAY_MS = 300000; // 5 minute delay on startup - - const job: ScheduledJob = { - name: JOB_NAME, - intervalId: null, - initialTimeoutId: null, - }; - - // Run after a longer delay on startup - job.initialTimeoutId = setTimeout(async () => { - try { - const result = await runSummaryGeneratorJob({ batchSize: 10 }); - if (result.summariesGenerated > 0) { - logger.info(result, 'Summary generator: initial run completed'); - } - } catch (err) { - logger.error({ err }, 'Summary generator: initial run failed'); - } - }, INITIAL_DELAY_MS); + shouldLogResult?: (result: TResult) => boolean; +} - // Then run periodically - job.intervalId = setInterval(async () => { - try { - const result = await runSummaryGeneratorJob({ batchSize: 10 }); - if (result.summariesGenerated > 0) { - logger.info(result, 'Summary generator: job completed'); - } - } catch (err) { - logger.error({ err }, 'Summary generator: job failed'); - } - }, INTERVAL_HOURS * 60 * 60 * 1000); +/** + * Runtime state for a running job + */ +interface RunningJob { + name: string; + intervalId: NodeJS.Timeout | null; + initialTimeoutId: NodeJS.Timeout | null; +} - this.jobs.set(JOB_NAME, job); - logger.debug({ intervalHours: INTERVAL_HOURS }, 'Summary generator job started'); +/** + * Convert a TimeInterval to milliseconds + */ +function toMilliseconds(interval: TimeInterval): number { + switch (interval.unit) { + case 'seconds': + return interval.value * 1000; + case 'minutes': + return interval.value * 60 * 1000; + case 'hours': + return interval.value * 60 * 60 * 1000; } +} - /** - * Start the account enrichment job - * Enriches accounts missing company data via Lusha API - * Prioritizes accounts with users over empty prospects - */ - startEnrichment(): void { - const JOB_NAME = 'account-enrichment'; - const INTERVAL_HOURS = 6; // Run every 6 hours - const INITIAL_DELAY_MS = 180000; // 3 minute delay on startup - - const job: ScheduledJob = { - name: JOB_NAME, - intervalId: null, - initialTimeoutId: null, - }; - - // Run after a delay on startup - job.initialTimeoutId = setTimeout(async () => { - try { - const result = await enrichMissingOrganizations({ - limit: 50, - includeEmptyProspects: true, - }); - if (result.enriched > 0 || result.failed > 0) { - logger.info(result, 'Account enrichment: initial run completed'); - } - } catch (err) { - logger.error({ err }, 'Account enrichment: initial run failed'); - } - }, INITIAL_DELAY_MS); - - // Then run periodically - job.intervalId = setInterval(async () => { - try { - const result = await enrichMissingOrganizations({ - limit: 50, - includeEmptyProspects: true, - }); - if (result.enriched > 0 || result.failed > 0) { - logger.info(result, 'Account enrichment: job completed'); - } - } catch (err) { - logger.error({ err }, 'Account enrichment: job failed'); - } - }, INTERVAL_HOURS * 60 * 60 * 1000); - - this.jobs.set(JOB_NAME, job); - logger.debug({ intervalHours: INTERVAL_HOURS }, 'Account enrichment job started'); +/** + * Check if current time is within business hours constraint (ET timezone) + */ +function isWithinBusinessHours(constraint: BusinessHoursConstraint): boolean { + const now = new Date(); + + // Get current hour in ET + const etHour = parseInt( + now.toLocaleString('en-US', { + timeZone: 'America/New_York', + hour: 'numeric', + hour12: false, + }), + 10 + ); + + // Check hour range + if (etHour < constraint.startHour || etHour >= constraint.endHour) { + return false; } - /** - * Stop the document indexer job - */ - stopDocumentIndexer(): void { - this.stopJob('document-indexer'); + // Check weekends if enabled (default: true) + if (constraint.skipWeekends !== false) { + const etDay = now.toLocaleString('en-US', { + timeZone: 'America/New_York', + weekday: 'short', + }); + if (etDay === 'Sat' || etDay === 'Sun') { + return false; + } } - /** - * Stop the proactive outreach job - */ - stopOutreach(): void { - this.stopJob('proactive-outreach'); - } + return true; +} - /** - * Stop the summary generator job - */ - stopSummaryGenerator(): void { - this.stopJob('summary-generator'); - } +/** + * Job Scheduler - manages registration and execution of scheduled jobs + */ +class JobScheduler { + private configs: Map = new Map(); + private runningJobs: Map = new Map(); /** - * Stop the account enrichment job + * Register a job configuration */ - stopEnrichment(): void { - this.stopJob('account-enrichment'); + register(config: JobConfig): void { + if (this.configs.has(config.name)) { + logger.warn({ jobName: config.name }, 'Job already registered, replacing'); + } + // Cast necessary because Map cannot hold heterogeneous generics. + // Type safety is enforced at registration time via register. + this.configs.set(config.name, config as JobConfig); } /** - * Start the Moltbook poster job - * Posts high-quality industry articles to Moltbook with Addie's take + * Start a specific job by name */ - startMoltbookPoster(): void { - const JOB_NAME = 'moltbook-poster'; - const INTERVAL_HOURS = 2; // Post every 2 hours - const INITIAL_DELAY_MS = 600000; // 10 minute delay on startup - - const job: ScheduledJob = { - name: JOB_NAME, - intervalId: null, - initialTimeoutId: null, - }; - - // Run after a delay on startup - job.initialTimeoutId = setTimeout(async () => { - try { - const result = await runMoltbookPosterJob({ limit: 1 }); - if (result.postsCreated > 0) { - logger.info(result, 'Moltbook poster: initial run completed'); - } - } catch (err) { - logger.error({ err }, 'Moltbook poster: initial run failed'); - } - }, INITIAL_DELAY_MS); - - // Then run periodically - job.intervalId = setInterval(async () => { - try { - const result = await runMoltbookPosterJob({ limit: 1 }); - if (result.postsCreated > 0) { - logger.info(result, 'Moltbook poster: job completed'); - } - } catch (err) { - logger.error({ err }, 'Moltbook poster: job failed'); - } - }, INTERVAL_HOURS * 60 * 60 * 1000); - - this.jobs.set(JOB_NAME, job); - logger.debug({ intervalHours: INTERVAL_HOURS }, 'Moltbook poster job started'); - } + start(name: string): void { + const config = this.configs.get(name); + if (!config) { + logger.error({ jobName: name }, 'Cannot start unknown job'); + return; + } - /** - * Start the Moltbook engagement job - * Searches for advertising discussions and engages thoughtfully - */ - startMoltbookEngagement(): void { - const JOB_NAME = 'moltbook-engagement'; - const INTERVAL_HOURS = 1; // Engage every hour - const INITIAL_DELAY_MS = 600000; // 10 minute delay on startup + if (this.runningJobs.has(name)) { + logger.warn({ jobName: name }, 'Job already running'); + return; + } - const job: ScheduledJob = { - name: JOB_NAME, + const job: RunningJob = { + name, intervalId: null, initialTimeoutId: null, }; - // Run after a delay on startup - job.initialTimeoutId = setTimeout(async () => { - try { - const result = await runMoltbookEngagementJob({ limit: 3 }); - if (result.commentsCreated > 0 || result.interestingThreads > 0) { - logger.info(result, 'Moltbook engagement: initial run completed'); - } - } catch (err) { - logger.error({ err }, 'Moltbook engagement: initial run failed'); + const runJob = async () => { + // Check business hours constraint if configured + if (config.businessHours && !isWithinBusinessHours(config.businessHours)) { + logger.debug({ jobName: name }, 'Skipping - outside business hours'); + return; } - }, INITIAL_DELAY_MS); - // Then run periodically - job.intervalId = setInterval(async () => { try { - const result = await runMoltbookEngagementJob({ limit: 3 }); - if (result.commentsCreated > 0 || result.interestingThreads > 0) { - logger.info(result, 'Moltbook engagement: job completed'); + const result = await config.runner(config.options ?? ({} as never)); + + // Log based on shouldLogResult predicate or default to debug + const shouldLog = config.shouldLogResult?.(result) ?? false; + if (shouldLog) { + logger.info({ jobName: name, result }, `${config.description}: completed`); + } else { + logger.debug({ jobName: name, result }, `${config.description}: completed`); } } catch (err) { - logger.error({ err }, 'Moltbook engagement: job failed'); + logger.error({ err, jobName: name }, `${config.description}: failed`); } - }, INTERVAL_HOURS * 60 * 60 * 1000); + }; - this.jobs.set(JOB_NAME, job); - logger.debug({ intervalHours: INTERVAL_HOURS }, 'Moltbook engagement job started'); - } + // Schedule initial run with delay + const initialDelayMs = config.initialDelay ? toMilliseconds(config.initialDelay) : 0; - /** - * Stop the Moltbook poster job - */ - stopMoltbookPoster(): void { - this.stopJob('moltbook-poster'); - } + if (initialDelayMs > 0) { + job.initialTimeoutId = setTimeout(runJob, initialDelayMs); + } else { + // Run on next tick to not block startup + setImmediate(runJob); + } - /** - * Stop the Moltbook engagement job - */ - stopMoltbookEngagement(): void { - this.stopJob('moltbook-engagement'); + // Schedule periodic runs + job.intervalId = setInterval(runJob, toMilliseconds(config.interval)); + + this.runningJobs.set(name, job); + logger.debug( + { + jobName: name, + interval: config.interval, + initialDelay: config.initialDelay, + businessHours: config.businessHours, + }, + `${config.description} scheduled` + ); } /** * Stop a specific job by name */ - private stopJob(name: string): void { - const job = this.jobs.get(name); - if (!job) return; + stop(name: string): void { + const job = this.runningJobs.get(name); + if (!job) { + return; + } if (job.initialTimeoutId) { clearTimeout(job.initialTimeoutId); - job.initialTimeoutId = null; } - if (job.intervalId) { clearInterval(job.intervalId); - job.intervalId = null; } - this.jobs.delete(name); + this.runningJobs.delete(name); logger.info({ jobName: name }, 'Job stopped'); } + /** + * Start all registered jobs + */ + startAll(): void { + for (const name of this.configs.keys()) { + this.start(name); + } + logger.info({ jobCount: this.configs.size }, 'All jobs started'); + } + /** * Stop all running jobs */ stopAll(): void { - for (const name of this.jobs.keys()) { - this.stopJob(name); + for (const name of this.runningJobs.keys()) { + this.stop(name); } - logger.info('All scheduled jobs stopped'); + logger.info('All jobs stopped'); + } + + /** + * Check if a job is currently running + */ + isRunning(name: string): boolean { + return this.runningJobs.has(name); + } + + /** + * Get list of registered job names + */ + getRegisteredJobs(): string[] { + return Array.from(this.configs.keys()); } } diff --git a/server/src/http.ts b/server/src/http.ts index b74a358c32..25f19c82ec 100644 --- a/server/src/http.ts +++ b/server/src/http.ts @@ -52,8 +52,8 @@ import { createSlackRouter } from "./routes/slack.js"; import { createWebhooksRouter } from "./routes/webhooks.js"; import { createWorkOSWebhooksRouter } from "./routes/workos-webhooks.js"; import { createAdminSlackRouter, createAdminEmailRouter, createAdminFeedsRouter, createAdminNotificationChannelsRouter, createAdminUsersRouter, createAdminSettingsRouter } from "./routes/admin/index.js"; -import { processFeedsToFetch } from "./addie/services/feed-fetcher.js"; -import { processAlerts } from "./addie/services/industry-alerts.js"; +import { jobScheduler } from "./addie/jobs/scheduler.js"; +import { registerAllJobs, JOB_NAMES } from "./addie/jobs/job-definitions.js"; import { createBillingRouter } from "./routes/billing.js"; import { createPublicBillingRouter } from "./routes/billing-public.js"; import { createOrganizationsRouter } from "./routes/organizations.js"; @@ -66,15 +66,9 @@ import { createMemberProfileRouter, createAdminMemberProfileRouter } from "./rou import { createAgentOAuthRouter } from "./routes/agent-oauth.js"; import { sendWelcomeEmail, sendUserSignupEmail, emailDb } from "./notifications/email.js"; import { emailPrefsDb } from "./db/email-preferences-db.js"; -import { queuePerspectiveLink, processPendingResources, processRssPerspectives, processCommunityArticles } from "./addie/services/content-curator.js"; +import { queuePerspectiveLink } from "./addie/services/content-curator.js"; import { InsightsDatabase } from "./db/insights-db.js"; -import { sendCommunityReplies } from "./addie/services/community-articles.js"; -import { sendChannelMessage } from "./slack/client.js"; import { serveHtmlWithMetaTags } from "./utils/html-config.js"; -import { runTaskReminderJob } from "./addie/jobs/task-reminder.js"; -import { runEngagementScoringJob } from "./addie/jobs/engagement-scoring.js"; -import { runGoalFollowUpJob } from "./addie/jobs/goal-follow-up.js"; -import { jobScheduler } from "./addie/jobs/scheduler.js"; import { notifyJoinRequest, notifyMemberAdded, notifySubscriptionThankYou } from "./slack/org-group-dm.js"; const __filename = fileURLToPath(import.meta.url); @@ -386,14 +380,6 @@ export class HTTPServer { private publisherTracker: PublisherTracker; private propertiesService: PropertiesService; private adagentsManager: AdAgentsManager; - private contentCuratorIntervalId: NodeJS.Timeout | null = null; - private feedFetcherIntervalId: NodeJS.Timeout | null = null; - private feedFetcherInitialTimeoutId: NodeJS.Timeout | null = null; - private alertProcessorIntervalId: NodeJS.Timeout | null = null; - private alertProcessorInitialTimeoutId: NodeJS.Timeout | null = null; - private taskReminderIntervalId: NodeJS.Timeout | null = null; - private engagementScoringIntervalId: NodeJS.Timeout | null = null; - private goalFollowUpIntervalId: NodeJS.Timeout | null = null; constructor() { this.app = express(); @@ -6338,40 +6324,25 @@ Disallow: /api/admin/ this.crawler.startPeriodicCrawl(salesAgents, 60); // Crawl every 60 minutes } - // Start periodic content curator for Addie's knowledge base - // Process pending external resources (fetch content, generate summaries) - this.startContentCurator(); - - // Start industry feed monitoring - // Fetches RSS feeds, processes articles, sends Slack alerts - this.startIndustryMonitor(); - - // Start daily task reminder job - // Sends DMs to admins about overdue/upcoming tasks - this.startTaskReminders(); - - // Start engagement scoring job - // Updates user and org engagement scores periodically - this.startEngagementScoring(); - - // Start goal follow-up job - // Sends follow-up messages and reconciles goal outcomes - this.startGoalFollowUp(); - - // Start committee document jobs via scheduler - jobScheduler.startDocumentIndexer(); - jobScheduler.startSummaryGenerator(); - - // Start proactive outreach job - jobScheduler.startOutreach(); - - // Start account enrichment job - jobScheduler.startEnrichment(); - - // Start Moltbook jobs (if API key is configured) + // Register and start all scheduled jobs + registerAllJobs(); + + // Start most jobs + jobScheduler.start(JOB_NAMES.DOCUMENT_INDEXER); + jobScheduler.start(JOB_NAMES.SUMMARY_GENERATOR); + jobScheduler.start(JOB_NAMES.PROACTIVE_OUTREACH); + jobScheduler.start(JOB_NAMES.ACCOUNT_ENRICHMENT); + jobScheduler.start(JOB_NAMES.CONTENT_CURATOR); + jobScheduler.start(JOB_NAMES.FEED_FETCHER); + jobScheduler.start(JOB_NAMES.ALERT_PROCESSOR); + jobScheduler.start(JOB_NAMES.TASK_REMINDER); + jobScheduler.start(JOB_NAMES.ENGAGEMENT_SCORING); + jobScheduler.start(JOB_NAMES.GOAL_FOLLOW_UP); + + // Start Moltbook jobs only if API key is configured if (process.env.MOLTBOOK_API_KEY) { - jobScheduler.startMoltbookPoster(); - jobScheduler.startMoltbookEngagement(); + jobScheduler.start(JOB_NAMES.MOLTBOOK_POSTER); + jobScheduler.start(JOB_NAMES.MOLTBOOK_ENGAGEMENT); } this.server = this.app.listen(port, () => { @@ -6386,258 +6357,6 @@ Disallow: /api/admin/ this.setupShutdownHandlers(); } - /** - * Start periodic content curator for Addie's knowledge base - * Processes pending external resources (fetch content, generate AI summaries) - */ - private startContentCurator(): void { - const CURATOR_INTERVAL_MINUTES = 5; - - // Process on startup after a short delay - setTimeout(async () => { - try { - // Process manually queued resources - const result = await processPendingResources({ limit: 5 }); - if (result.processed > 0) { - logger.debug(result, 'Content curator: processed pending resources'); - } - // Process RSS perspectives - const rssResult = await processRssPerspectives({ limit: 5 }); - if (rssResult.processed > 0) { - logger.debug(rssResult, 'Content curator: processed RSS perspectives'); - } - // Process community articles - const communityResult = await processCommunityArticles({ limit: 5 }); - if (communityResult.processed > 0) { - logger.debug(communityResult, 'Content curator: processed community articles'); - } - // Send replies to processed community articles - const replyResult = await sendCommunityReplies(async (channelId, threadTs, text) => { - const result = await sendChannelMessage(channelId, { text, thread_ts: threadTs }); - return result.ok; - }); - if (replyResult.sent > 0) { - logger.debug(replyResult, 'Content curator: sent community article replies'); - } - } catch (err) { - logger.error({ err }, 'Content curator: initial processing failed'); - } - }, 30000); // 30 second delay to let other services start - - // Then process periodically - this.contentCuratorIntervalId = setInterval(async () => { - try { - // Process manually queued resources - const result = await processPendingResources({ limit: 5 }); - if (result.processed > 0) { - logger.debug(result, 'Content curator: processed pending resources'); - } - // Process RSS perspectives - const rssResult = await processRssPerspectives({ limit: 5 }); - if (rssResult.processed > 0) { - logger.debug(rssResult, 'Content curator: processed RSS perspectives'); - } - // Process community articles - const communityResult = await processCommunityArticles({ limit: 5 }); - if (communityResult.processed > 0) { - logger.debug(communityResult, 'Content curator: processed community articles'); - } - // Send replies to processed community articles - const replyResult = await sendCommunityReplies(async (channelId, threadTs, text) => { - const result = await sendChannelMessage(channelId, { text, thread_ts: threadTs }); - return result.ok; - }); - if (replyResult.sent > 0) { - logger.debug(replyResult, 'Content curator: sent community article replies'); - } - } catch (err) { - logger.error({ err }, 'Content curator: periodic processing failed'); - } - }, CURATOR_INTERVAL_MINUTES * 60 * 1000); - - logger.debug({ intervalMinutes: CURATOR_INTERVAL_MINUTES }, 'Content curator started'); - } - - /** - * Start industry feed monitoring system - * Fetches RSS feeds from ad tech publications, processes articles, - * and sends Slack alerts for high-priority content - */ - private startIndustryMonitor(): void { - const FEED_FETCH_INTERVAL_MINUTES = 30; - const ALERT_CHECK_INTERVAL_MINUTES = 5; - - // Feed fetcher - check feeds every 30 minutes - // Creates perspectives from RSS articles, which are then processed by the content curator - this.feedFetcherInitialTimeoutId = setTimeout(async () => { - this.feedFetcherInitialTimeoutId = null; - try { - const result = await processFeedsToFetch(); - if (result.feedsProcessed > 0) { - logger.debug(result, 'Industry monitor: fetched RSS feeds'); - } - } catch (err) { - logger.error({ err }, 'Industry monitor: initial feed fetch failed'); - } - }, 60000); // 1 minute delay to let other services start - - this.feedFetcherIntervalId = setInterval(async () => { - try { - const result = await processFeedsToFetch(); - if (result.feedsProcessed > 0) { - logger.debug(result, 'Industry monitor: fetched RSS feeds'); - } - } catch (err) { - logger.error({ err }, 'Industry monitor: periodic feed fetch failed'); - } - }, FEED_FETCH_INTERVAL_MINUTES * 60 * 1000); - - // Alert processor - check for alerts every 5 minutes - this.alertProcessorInitialTimeoutId = setTimeout(async () => { - this.alertProcessorInitialTimeoutId = null; - try { - const result = await processAlerts(); - if (result.alerted > 0) { - logger.info(result, 'Industry monitor: sent alerts'); - } - } catch (err) { - logger.error({ err }, 'Industry monitor: initial alert processing failed'); - } - }, 120000); // 2 minute delay - - this.alertProcessorIntervalId = setInterval(async () => { - try { - const result = await processAlerts(); - if (result.alerted > 0) { - logger.info(result, 'Industry monitor: sent alerts'); - } - } catch (err) { - logger.error({ err }, 'Industry monitor: periodic alert processing failed'); - } - }, ALERT_CHECK_INTERVAL_MINUTES * 60 * 1000); - - logger.debug({ - feedFetchIntervalMinutes: FEED_FETCH_INTERVAL_MINUTES, - alertCheckIntervalMinutes: ALERT_CHECK_INTERVAL_MINUTES, - }, 'Industry monitor started'); - } - - /** - * Start daily task reminder job - * Sends DMs to admins about overdue/upcoming tasks at 9am ET - */ - private startTaskReminders(): void { - const REMINDER_CHECK_INTERVAL_HOURS = 1; - - // Check every hour, but only send once per day per user - this.taskReminderIntervalId = setInterval(async () => { - try { - // Only run during morning hours (8-10am ET) using proper timezone handling - const now = new Date(); - const etHour = parseInt(now.toLocaleString('en-US', { - timeZone: 'America/New_York', - hour: 'numeric', - hour12: false, - }), 10); - if (etHour < 8 || etHour > 10) { - return; - } - - // Skip weekends (using ET day of week) - const etDayStr = now.toLocaleString('en-US', { - timeZone: 'America/New_York', - weekday: 'short', - }); - if (etDayStr === 'Sat' || etDayStr === 'Sun') { - return; - } - - const result = await runTaskReminderJob(); - if (result.remindersSent > 0) { - logger.info(result, 'Task reminders: sent daily reminders'); - } - } catch (err) { - logger.error({ err }, 'Task reminders: job failed'); - } - }, REMINDER_CHECK_INTERVAL_HOURS * 60 * 60 * 1000); - - logger.debug({ intervalHours: REMINDER_CHECK_INTERVAL_HOURS }, 'Task reminder job started'); - } - - /** - * Start periodic engagement scoring for users and organizations - * Updates stale scores (older than 1 day) every hour - */ - private startEngagementScoring(): void { - const SCORING_INTERVAL_HOURS = 1; - - // Run immediately on startup (with delay for DB connection) - setTimeout(async () => { - try { - await runEngagementScoringJob(); - } catch (err) { - logger.error({ err }, 'Engagement scoring: initial batch failed'); - } - }, 10000); - - // Then run periodically - this.engagementScoringIntervalId = setInterval(async () => { - try { - await runEngagementScoringJob(); - } catch (err) { - logger.error({ err }, 'Engagement scoring: job failed'); - } - }, SCORING_INTERVAL_HOURS * 60 * 60 * 1000); - - logger.debug({ intervalHours: SCORING_INTERVAL_HOURS }, 'Engagement scoring job started'); - } - - /** - * Start periodic goal follow-up job - * Sends follow-up messages for unanswered outreach and reconciles goal outcomes - */ - private startGoalFollowUp(): void { - const FOLLOW_UP_INTERVAL_HOURS = 4; // Check every 4 hours - - // Run after a delay on startup - setTimeout(async () => { - try { - await runGoalFollowUpJob(); - } catch (err) { - logger.error({ err }, 'Goal follow-up: initial run failed'); - } - }, 180000); // 3 minute delay - - // Then run periodically - this.goalFollowUpIntervalId = setInterval(async () => { - try { - // Only run during business hours (9am-6pm ET) - const now = new Date(); - const etHour = parseInt(now.toLocaleString('en-US', { - timeZone: 'America/New_York', - hour: 'numeric', - hour12: false, - }), 10); - const etDay = now.toLocaleString('en-US', { - timeZone: 'America/New_York', - weekday: 'short', - }); - - // Skip weekends and outside business hours - if (['Sat', 'Sun'].includes(etDay) || etHour < 9 || etHour >= 18) { - logger.debug({ etHour, etDay }, 'Goal follow-up: skipping outside business hours'); - return; - } - - await runGoalFollowUpJob(); - } catch (err) { - logger.error({ err }, 'Goal follow-up: job failed'); - } - }, FOLLOW_UP_INTERVAL_HOURS * 60 * 60 * 1000); - - logger.debug({ intervalHours: FOLLOW_UP_INTERVAL_HOURS }, 'Goal follow-up job started'); - } - /** * Setup graceful shutdown handlers for SIGTERM and SIGINT */ @@ -6658,54 +6377,7 @@ Disallow: /api/admin/ async stop(): Promise { logger.info('Stopping HTTP server'); - // Stop content curator - if (this.contentCuratorIntervalId) { - clearInterval(this.contentCuratorIntervalId); - this.contentCuratorIntervalId = null; - logger.info('Content curator stopped'); - } - - // Stop industry monitor jobs - if (this.feedFetcherInitialTimeoutId) { - clearTimeout(this.feedFetcherInitialTimeoutId); - this.feedFetcherInitialTimeoutId = null; - } - if (this.feedFetcherIntervalId) { - clearInterval(this.feedFetcherIntervalId); - this.feedFetcherIntervalId = null; - } - if (this.alertProcessorInitialTimeoutId) { - clearTimeout(this.alertProcessorInitialTimeoutId); - this.alertProcessorInitialTimeoutId = null; - } - if (this.alertProcessorIntervalId) { - clearInterval(this.alertProcessorIntervalId); - this.alertProcessorIntervalId = null; - } - logger.info('Industry monitor stopped'); - - // Stop task reminder job - if (this.taskReminderIntervalId) { - clearInterval(this.taskReminderIntervalId); - this.taskReminderIntervalId = null; - logger.info('Task reminder job stopped'); - } - - // Stop engagement scoring job - if (this.engagementScoringIntervalId) { - clearInterval(this.engagementScoringIntervalId); - this.engagementScoringIntervalId = null; - logger.info('Engagement scoring job stopped'); - } - - // Stop goal follow-up job - if (this.goalFollowUpIntervalId) { - clearInterval(this.goalFollowUpIntervalId); - this.goalFollowUpIntervalId = null; - logger.info('Goal follow-up job stopped'); - } - - // Stop committee document jobs via scheduler + // Stop all scheduled jobs jobScheduler.stopAll(); // Close HTTP server