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
This commit is contained in:
Christomatt
2026-01-26 09:39:12 +01:00
parent 731838dd48
commit e98221e81a
6 changed files with 354 additions and 193 deletions
+10 -16
View File
@@ -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) {
+6 -1
View File
@@ -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 });
}
+13 -23
View File
@@ -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}`);
+166 -108
View File
@@ -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<boolean> {
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<AutonomousPostEvaluation> {
// 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,
};
+89 -45
View File
@@ -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<ProcessScheduledPostsResu
continue;
}
// Fetch fresh content from sources before checking availability
await fetchAllSourcesForBot(bot.id, { maxItems: 20, timeout: 15000 });
// Check if bot has content sources
const hasContentSources = await hasUnprocessedContent(bot.id).catch(() => 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<ProcessScheduledPostsResu
return result;
}
/**
* Check if a bot has any active content sources.
*
* @param botId - The bot ID
* @returns True if bot has active sources
*/
async function botHasActiveSources(botId: string): Promise<boolean> {
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.
*
+70
View File
@@ -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
*/