From e98221e81af3bf14c9feb16621ae30e08e8a3434 Mon Sep 17 00:00:00 2001 From: Christomatt Date: Mon, 26 Jan 2026 09:39:12 +0100 Subject: [PATCH] feat(bots): Consolidate bot posting into autonomous module and improve caching - Remove scheduled posting logic and consolidate all bot posting into autonomous module - Simplify cron endpoint to only call processAllAutonomousBots with cleaner response format - Add background post caching for swarm follow operations via cacheSwarmUserPosts - Update background scheduler to remove scheduled posting references and streamline logging - Improve autonomous module documentation to clarify schedule-based posting with optional sources - Enhance error tracking and reporting across bot task execution - Reduce code duplication by eliminating separate scheduled/autonomous processing paths --- src/app/api/cron/bots/route.ts | 26 +- src/app/api/users/[handle]/follow/route.ts | 7 +- src/lib/background/scheduler.ts | 36 +-- src/lib/bots/autonomous.ts | 274 +++++++++++++-------- src/lib/bots/scheduler.ts | 134 ++++++---- src/lib/swarm/interactions.ts | 70 ++++++ 6 files changed, 354 insertions(+), 193 deletions(-) diff --git a/src/app/api/cron/bots/route.ts b/src/app/api/cron/bots/route.ts index 1768bce..2d47d49 100644 --- a/src/app/api/cron/bots/route.ts +++ b/src/app/api/cron/bots/route.ts @@ -1,11 +1,10 @@ /** - * Cron endpoint for bot scheduled posting + * Cron endpoint for bot autonomous posting * * Call this endpoint periodically (e.g., every minute) via cron job or PM2 */ import { NextRequest, NextResponse } from 'next/server'; -import { processScheduledPosts } from '@/lib/bots/scheduler'; import { processAllAutonomousBots } from '@/lib/bots/autonomous'; export async function POST(request: NextRequest) { @@ -18,23 +17,18 @@ export async function POST(request: NextRequest) { } try { - // Process scheduled posts - const scheduledResult = await processScheduledPosts(); - - // Process autonomous bots - const autonomousResult = await processAllAutonomousBots(); + const results = await processAllAutonomousBots(); return NextResponse.json({ success: true, - scheduled: { - processed: scheduledResult.processed, - skipped: scheduledResult.skipped, - errors: scheduledResult.errors.length, - }, - autonomous: { - total: autonomousResult.length, - posted: autonomousResult.filter(r => r.result.posted).length, - }, + total: results.length, + posted: results.filter(r => r.result.posted).length, + errors: results.filter(r => r.error).length, + details: results.map(r => ({ + handle: r.botHandle, + posted: r.result.posted, + reason: r.result.reason || r.error, + })), timestamp: new Date().toISOString(), }); } catch (error) { diff --git a/src/app/api/users/[handle]/follow/route.ts b/src/app/api/users/[handle]/follow/route.ts index 1548f95..1705e6d 100644 --- a/src/app/api/users/[handle]/follow/route.ts +++ b/src/app/api/users/[handle]/follow/route.ts @@ -7,7 +7,7 @@ import { resolveRemoteUser } from '@/lib/activitypub/fetch'; import { createFollowActivity, createUndoActivity } from '@/lib/activitypub/activities'; import { deliverActivity } from '@/lib/activitypub/outbox'; import { cacheRemoteUserPosts } from '@/lib/activitypub/cache'; -import { isSwarmNode, deliverSwarmFollow, deliverSwarmUnfollow } from '@/lib/swarm/interactions'; +import { isSwarmNode, deliverSwarmFollow, deliverSwarmUnfollow, cacheSwarmUserPosts } from '@/lib/swarm/interactions'; type RouteContext = { params: Promise<{ handle: string }> }; @@ -156,6 +156,11 @@ export async function POST(request: Request, context: RouteContext) { .set({ followingCount: currentUser.followingCount + 1 }) .where(eq(users.id, currentUser.id)); + // Cache the remote user's recent posts in the background + cacheSwarmUserPosts(remote.handle, remote.domain, targetHandle, 20) + .then(result => console.log(`[Swarm] Cached ${result.cached} posts for ${targetHandle}`)) + .catch(err => console.error('[Swarm] Error caching remote posts:', err)); + console.log(`[Swarm] Follow delivered to ${remote.domain} for @${remote.handle}`); return NextResponse.json({ success: true, following: true, remote: true, swarm: true }); } diff --git a/src/lib/background/scheduler.ts b/src/lib/background/scheduler.ts index 4397138..e24d889 100644 --- a/src/lib/background/scheduler.ts +++ b/src/lib/background/scheduler.ts @@ -2,12 +2,11 @@ * Background Task Scheduler * * Runs periodic tasks within the Next.js process: - * - Bot scheduling (every 1 minute) + * - Bot autonomous posting (every 1 minute) * - Swarm gossip (every 5 minutes) * - Swarm announcement (on startup) */ -import { processScheduledPosts } from '@/lib/bots/scheduler'; import { processAllAutonomousBots } from '@/lib/bots/autonomous'; import { runGossipRound } from '@/lib/swarm/gossip'; import { announceToSeeds } from '@/lib/swarm/discovery'; @@ -30,39 +29,30 @@ function log(category: string, message: string, data?: unknown) { async function runBotTasks() { try { - const scheduledResult = await processScheduledPosts(); - const autonomousResult = await processAllAutonomousBots(); + const results = await processAllAutonomousBots(); - const posted = autonomousResult.filter(r => r.result.posted).length; - const skipped = scheduledResult.skipped; - const errors = scheduledResult.errors.length + autonomousResult.filter(r => r.error).length; + const posted = results.filter(r => r.result.posted).length; + const errors = results.filter(r => r.error).length; - // Always log bot task results for debugging - if (scheduledResult.processed > 0 || posted > 0) { - log('BOTS', `Processed ${scheduledResult.processed} scheduled, ${posted} autonomous posts`); - } else if (scheduledResult.details.length > 0 || autonomousResult.length > 0) { + if (posted > 0) { + log('BOTS', `Created ${posted} posts`); + } else if (results.length > 0) { // Log why bots didn't post - const reasons = scheduledResult.details - .filter(d => d.status !== 'posted') - .map(d => `${d.botId.slice(0, 8)}: ${d.status}${d.message ? ` (${d.message})` : ''}`) - .slice(0, 3); - - const autoReasons = autonomousResult + const reasons = results .filter(r => !r.result.posted) .map(r => `${r.botHandle}: ${r.result.reason || r.error || 'unknown'}`) - .slice(0, 3); + .slice(0, 5); - if (reasons.length > 0 || autoReasons.length > 0) { - log('BOTS', `No posts created. Scheduled: ${scheduledResult.details.length} checked, ${skipped} skipped. Autonomous: ${autonomousResult.length} checked.`); - if (reasons.length > 0) log('BOTS', `Scheduled skip reasons: ${reasons.join('; ')}`); - if (autoReasons.length > 0) log('BOTS', `Autonomous skip reasons: ${autoReasons.join('; ')}`); + if (reasons.length > 0) { + log('BOTS', `${results.length} bots checked, no posts. Reasons: ${reasons.join('; ')}`); } } else { log('BOTS', 'No active bots found'); } if (errors > 0) { - log('BOTS', `Errors: ${scheduledResult.errors.join('; ')}`); + const errorMsgs = results.filter(r => r.error).map(r => `${r.botHandle}: ${r.error}`); + log('BOTS', `Errors: ${errorMsgs.join('; ')}`); } } catch (error) { log('BOTS', `Error: ${error}`); diff --git a/src/lib/bots/autonomous.ts b/src/lib/bots/autonomous.ts index cc15fca..c5ca18f 100644 --- a/src/lib/bots/autonomous.ts +++ b/src/lib/bots/autonomous.ts @@ -2,8 +2,8 @@ * Autonomous Posting Module * * Implements autonomous posting logic for bots. When autonomous mode is enabled, - * bots evaluate content from their sources and decide whether to post based on - * content interest. All posting respects rate limits. + * bots post based on their schedule configuration. Content sources are optional - + * bots without sources generate original posts based on their personality. * * Requirements: 6.1, 6.2, 6.3, 6.5 */ @@ -12,8 +12,8 @@ import { db, botContentItems, bots } from '@/db'; import { eq, and } from 'drizzle-orm'; import { ContentGenerator, type Bot as ContentGeneratorBot, type ContentItem } from './contentGenerator'; import { canPost, recordPost } from './rateLimiter'; -import { getBotById } from './botManager'; import { decryptApiKey, deserializeEncryptedData } from './encryption'; +import { parseScheduleConfig, isDue } from './scheduler'; // ============================================ // TYPES @@ -105,6 +105,25 @@ export const MIN_INTEREST_SCORE = 60; // HELPER FUNCTIONS // ============================================ +/** + * Check if a bot has any active content sources configured. + * + * @param botId - The ID of the bot + * @returns True if bot has content sources + */ +async function hasContentSources(botId: string): Promise { + const bot = await db.query.bots.findFirst({ + where: eq(bots.id, botId), + with: { + contentSources: { + where: (sources, { eq }) => eq(sources.isActive, true), + }, + }, + }); + + return !!(bot?.contentSources && bot.contentSources.length > 0); +} + /** * Get unprocessed content items for a bot. * Returns content items that haven't been processed yet, ordered by published date. @@ -261,25 +280,7 @@ export async function evaluateContentForPosting( botId: string, contentItem: ContentItem ): Promise { - // Get bot - const bot = await getBotById(botId); - if (!bot) { - throw new AutonomousPostError( - `Bot not found: ${botId}`, - 'BOT_NOT_FOUND' - ); - } - - // Check if autonomous mode is enabled - if (!bot.autonomousMode) { - return { - shouldPost: false, - reason: 'Autonomous mode is disabled for this bot', - contentItem, - }; - } - - // Get bot with encrypted API key + // Get bot with user relation const dbBot = await db.query.bots.findFirst({ where: eq(bots.id, botId), with: { user: true }, @@ -292,6 +293,15 @@ export async function evaluateContentForPosting( ); } + // Check if autonomous mode is enabled + if (!dbBot.autonomousMode) { + return { + shouldPost: false, + reason: 'Autonomous mode is disabled for this bot', + contentItem, + }; + } + // Check if bot has API key try { const apiKey = getDecryptedApiKeyForBot(dbBot); @@ -340,7 +350,7 @@ export async function evaluateContentForPosting( /** * Attempt to create an autonomous post for a bot. - * Evaluates unprocessed content and posts if interesting and rate limits allow. + * Checks schedule timing, then either uses content sources or generates original posts. * * This is the main entry point for autonomous posting. * @@ -359,9 +369,13 @@ export async function attemptAutonomousPost( skipRateLimitCheck = false, } = options; - // Get bot - const bot = await getBotById(botId); - if (!bot) { + // Get bot with schedule config + const dbBot = await db.query.bots.findFirst({ + where: eq(bots.id, botId), + with: { user: true }, + }); + + if (!dbBot) { throw new AutonomousPostError( `Bot not found: ${botId}`, 'BOT_NOT_FOUND' @@ -369,13 +383,40 @@ export async function attemptAutonomousPost( } // Check if autonomous mode is enabled (Requirement 6.5) - if (!bot.autonomousMode) { + if (!dbBot.autonomousMode) { return { posted: false, - reason: 'Autonomous mode is disabled for this bot', + reason: 'Autonomous mode is disabled', }; } + // Check if bot is active and not suspended + if (!dbBot.isActive) { + return { + posted: false, + reason: 'Bot is not active', + }; + } + + if (dbBot.isSuspended) { + return { + posted: false, + reason: 'Bot is suspended', + }; + } + + // Check schedule timing (if configured) + const scheduleConfig = parseScheduleConfig(dbBot.scheduleConfig); + if (scheduleConfig) { + const dueResult = isDue(scheduleConfig, dbBot.lastPostAt); + if (!dueResult.isDue) { + return { + posted: false, + reason: dueResult.reason || 'Not scheduled to post yet', + }; + } + } + // Check rate limits (Requirement 6.3) if (!skipRateLimitCheck) { const rateLimitCheck = await canPost(botId); @@ -387,95 +428,105 @@ export async function attemptAutonomousPost( } } - // Get unprocessed content items (Requirement 6.1) - const contentItems = await getUnprocessedContentItems(botId, maxEvaluations); - - if (contentItems.length === 0) { - return { - posted: false, - reason: 'No unprocessed content available', - }; + // Check if bot has API key + try { + const apiKey = getDecryptedApiKeyForBot(dbBot); + if (!apiKey || apiKey === '__REMOVED__') { + throw new AutonomousPostError( + 'Bot does not have a valid API key configured', + 'NO_API_KEY' + ); + } + } catch (error) { + if (error instanceof AutonomousPostError) throw error; + throw new AutonomousPostError( + 'Failed to decrypt bot API key', + 'NO_API_KEY', + error instanceof Error ? error : undefined + ); } - // Evaluate content items until we find one worth posting (Requirement 6.2) - for (const contentItem of contentItems) { - try { - const evaluation = await evaluateContentForPosting(botId, contentItem); + const contentGeneratorBot = toContentGeneratorBot(dbBot, dbBot.user.handle); + const generator = new ContentGenerator(contentGeneratorBot); - // Mark as processed regardless of decision - await markContentItemProcessed( - contentItem.id, - undefined, - evaluation.interestScore, - evaluation.reason - ); + // Check if bot has content sources + const hasSourcesConfigured = await hasContentSources(botId); - if (evaluation.shouldPost) { - // Generate and create the post (Requirement 6.3) + if (hasSourcesConfigured) { + // Bot has content sources - try content-based posting first + const contentItems = await getUnprocessedContentItems(botId, maxEvaluations); + + if (contentItems.length > 0) { + // Evaluate content items until we find one worth posting + for (const contentItem of contentItems) { try { - const dbBot = await db.query.bots.findFirst({ - where: eq(bots.id, botId), - with: { user: true }, - }); + const evaluation = await evaluateContentForPosting(botId, contentItem); - if (!dbBot) { - throw new Error('Bot not found'); - } - - const contentGeneratorBot = toContentGeneratorBot(dbBot, dbBot.user.handle); - const generator = new ContentGenerator(contentGeneratorBot); - - const generatedContent = await generator.generatePost(contentItem); - - // Record the post for rate limiting - if (!skipRateLimitCheck) { - await recordPost(botId); - } - - // Update the content item with the post ID - // Note: Actual post creation would happen in the posting module - // For now, we just return the generated content + // Mark as processed regardless of decision await markContentItemProcessed( contentItem.id, - 'pending', // Placeholder - actual post ID would be set by posting module + undefined, evaluation.interestScore, evaluation.reason ); - return { - posted: true, - postText: generatedContent.text, - contentItem, - }; + if (evaluation.shouldPost) { + const generatedContent = await generator.generatePost(contentItem); + + // Record the post for rate limiting + if (!skipRateLimitCheck) { + await recordPost(botId); + } + + await markContentItemProcessed( + contentItem.id, + 'pending', + evaluation.interestScore, + evaluation.reason + ); + + return { + posted: true, + postText: generatedContent.text, + contentItem, + }; + } } catch (error) { - throw new AutonomousPostError( - `Failed to create post: ${error instanceof Error ? error.message : String(error)}`, - 'POST_CREATION_FAILED', - error instanceof Error ? error : undefined + console.error(`Error evaluating content item ${contentItem.id}:`, error); + await markContentItemProcessed( + contentItem.id, + undefined, + 0, + `Evaluation failed: ${error instanceof Error ? error.message : String(error)}` ); + continue; } } - } catch (error) { - // Log error but continue to next content item - console.error(`Error evaluating content item ${contentItem.id}:`, error); - - // Mark as processed with error - await markContentItemProcessed( - contentItem.id, - undefined, - 0, - `Evaluation failed: ${error instanceof Error ? error.message : String(error)}` - ); - - continue; + // No interesting content found - fall through to generate original post } } - // No interesting content found - return { - posted: false, - reason: 'No interesting content found after evaluating available items', - }; + // Generate original post based on personality (no content source needed) + try { + const generatedContent = await generator.generatePost(undefined, undefined); + + // Record the post for rate limiting + if (!skipRateLimitCheck) { + await recordPost(botId); + } + + return { + posted: true, + postText: generatedContent.text, + }; + } catch (error) { + if (error instanceof AutonomousPostError) throw error; + throw new AutonomousPostError( + `Failed to create post: ${error instanceof Error ? error.message : String(error)}`, + 'POST_CREATION_FAILED', + error instanceof Error ? error : undefined + ); + } } /** @@ -542,7 +593,10 @@ export async function canPostAutonomously(botId: string): Promise<{ reason?: string; }> { // Get bot - const bot = await getBotById(botId); + const bot = await db.query.bots.findFirst({ + where: eq(bots.id, botId), + }); + if (!bot) { return { canPost: false, @@ -573,6 +627,18 @@ export async function canPostAutonomously(botId: string): Promise<{ }; } + // Check schedule timing (if configured) + const scheduleConfig = parseScheduleConfig(bot.scheduleConfig); + if (scheduleConfig) { + const dueResult = isDue(scheduleConfig, bot.lastPostAt); + if (!dueResult.isDue) { + return { + canPost: false, + reason: dueResult.reason || 'Not scheduled to post yet', + }; + } + } + // Check rate limits (Requirement 6.3) const rateLimitCheck = await canPost(botId); if (!rateLimitCheck.allowed) { @@ -582,15 +648,7 @@ export async function canPostAutonomously(botId: string): Promise<{ }; } - // Check if there's content to evaluate (Requirement 6.1) - const contentItems = await getUnprocessedContentItems(botId, 1); - if (contentItems.length === 0) { - return { - canPost: false, - reason: 'No unprocessed content available', - }; - } - + // Bots can always post - content sources are optional return { canPost: true, }; diff --git a/src/lib/bots/scheduler.ts b/src/lib/bots/scheduler.ts index 8d176d4..4c674e9 100644 --- a/src/lib/bots/scheduler.ts +++ b/src/lib/bots/scheduler.ts @@ -773,6 +773,7 @@ export async function getNextUnprocessedContent(botId: string): Promise<{ /** * Process scheduled posts for all active bots. * Checks each bot's schedule, rate limits, and content availability. + * Bots without content sources can still post based on their personality. * * @returns Processing result with statistics * @@ -835,54 +836,79 @@ export async function processScheduledPosts(): Promise false); - // Check for unprocessed content (Requirement 5.5) - const hasContent = await hasUnprocessedContent(bot.id); - - if (!hasContent) { - result.details.push({ - botId: bot.id, - status: 'skipped_no_content', - message: 'No unprocessed content available', + // If bot has content sources, fetch fresh content and check availability + if (hasContentSources || await botHasActiveSources(bot.id)) { + // Fetch fresh content from sources before checking availability + await fetchAllSourcesForBot(bot.id, { maxItems: 20, timeout: 15000 }); + + // Check for unprocessed content (Requirement 5.5) + const hasContent = await hasUnprocessedContent(bot.id); + + if (!hasContent) { + result.details.push({ + botId: bot.id, + status: 'skipped_no_content', + message: 'No unprocessed content available', + }); + result.skipped++; + continue; + } + + // Get the next content item + const contentItem = await getNextUnprocessedContent(bot.id); + + if (!contentItem) { + result.details.push({ + botId: bot.id, + status: 'skipped_no_content', + message: 'Failed to retrieve content item', + }); + result.skipped++; + continue; + } + + // Trigger post creation with the content item + const postResult = await triggerPost(bot.id, { + sourceContentId: contentItem.id, }); - result.skipped++; - continue; - } - - // Get the next content item - const contentItem = await getNextUnprocessedContent(bot.id); - - if (!contentItem) { - result.details.push({ - botId: bot.id, - status: 'skipped_no_content', - message: 'Failed to retrieve content item', - }); - result.skipped++; - continue; - } - - // Trigger post creation with the content item - const postResult = await triggerPost(bot.id, { - sourceContentId: contentItem.id, - }); - - if (postResult.success) { - result.details.push({ - botId: bot.id, - status: 'posted', - message: `Posted: ${contentItem.title.substring(0, 50)}...`, - }); - result.processed++; + + if (postResult.success) { + result.details.push({ + botId: bot.id, + status: 'posted', + message: `Posted: ${contentItem.title.substring(0, 50)}...`, + }); + result.processed++; + } else { + result.details.push({ + botId: bot.id, + status: 'error', + message: postResult.error || 'Failed to create post', + }); + result.errors.push(`Bot ${bot.id}: ${postResult.error}`); + } } else { - result.details.push({ - botId: bot.id, - status: 'error', - message: postResult.error || 'Failed to create post', - }); - result.errors.push(`Bot ${bot.id}: ${postResult.error}`); + // Bot has no content sources - generate original post based on personality + const postResult = await triggerPost(bot.id, {}); + + if (postResult.success) { + result.details.push({ + botId: bot.id, + status: 'posted', + message: 'Posted original content', + }); + result.processed++; + } else { + result.details.push({ + botId: bot.id, + status: 'error', + message: postResult.error || 'Failed to create post', + }); + result.errors.push(`Bot ${bot.id}: ${postResult.error}`); + } } } catch (error) { @@ -899,6 +925,24 @@ export async function processScheduledPosts(): Promise { + const sources = await db.query.botContentSources.findMany({ + where: and( + eq(botContentSources.botId, botId), + eq(botContentSources.isActive, true) + ), + columns: { id: true }, + }); + + return sources.length > 0; +} + /** * Get the schedule configuration for a bot. * diff --git a/src/lib/swarm/interactions.ts b/src/lib/swarm/interactions.ts index 6b72815..2632050 100644 --- a/src/lib/swarm/interactions.ts +++ b/src/lib/swarm/interactions.ts @@ -362,6 +362,76 @@ export async function fetchSwarmUserProfile( } } +/** + * Cache swarm user posts in the remotePosts table + * Similar to cacheRemoteUserPosts but for swarm nodes + */ +export async function cacheSwarmUserPosts( + handle: string, + domain: string, + fullHandle: string, // e.g., "user@domain.com" + limit: number = 20 +): Promise<{ cached: number; skipped: number }> { + try { + const profileData = await fetchSwarmUserProfile(handle, domain, limit); + + if (!profileData || !profileData.posts) { + return { cached: 0, skipped: 0 }; + } + + const { db, remotePosts } = await import('@/db'); + const { eq } = await import('drizzle-orm'); + + if (!db) { + return { cached: 0, skipped: 0 }; + } + + let cached = 0; + let skipped = 0; + + const actorUrl = `swarm://${domain}/${handle}`; + const profile = profileData.profile; + + for (const post of profileData.posts) { + // Generate a unique AP-style ID for the post + const apId = `swarm://${domain}/posts/${post.id}`; + + // Check if we already have this post + const existing = await db.query.remotePosts.findFirst({ + where: eq(remotePosts.apId, apId), + }); + + if (existing) { + skipped++; + continue; + } + + // Cache the post + await db.insert(remotePosts).values({ + apId, + authorHandle: fullHandle, + authorActorUrl: actorUrl, + authorDisplayName: profile.displayName || handle, + authorAvatarUrl: profile.avatarUrl || null, + content: post.content, + publishedAt: new Date(post.createdAt), + linkPreviewUrl: post.linkPreviewUrl || null, + linkPreviewTitle: post.linkPreviewTitle || null, + linkPreviewDescription: post.linkPreviewDescription || null, + linkPreviewImage: post.linkPreviewImage || null, + mediaJson: post.media ? JSON.stringify(post.media) : null, + }); + + cached++; + } + + return { cached, skipped }; + } catch (error) { + console.error(`[Swarm] Error caching posts for ${fullHandle}:`, error); + return { cached: 0, skipped: 0 }; + } +} + /** * Fetch a single post from a swarm node */