diff --git a/src/app/api/posts/route.ts b/src/app/api/posts/route.ts index 0a79060..a7e32f7 100644 --- a/src/app/api/posts/route.ts +++ b/src/app/api/posts/route.ts @@ -519,23 +519,92 @@ export async function GET(request: Request) { const followedRemoteUsers = await db.query.remoteFollows.findMany({ where: eq(remoteFollows.followerId, user.id), }); - const followedRemoteHandles = followedRemoteUsers.map(f => f.targetHandle); - // Get cached remote posts from followed users - let remotePostsData: typeof remotePosts.$inferSelect[] = []; - if (followedRemoteHandles.length > 0) { - remotePostsData = await db.query.remotePosts.findMany({ - where: inArray(remotePosts.authorHandle, followedRemoteHandles), - orderBy: [desc(remotePosts.publishedAt)], - limit: limit, + // Fetch posts LIVE from followed remote users (in parallel, with timeout) + let liveRemotePosts: any[] = []; + if (followedRemoteUsers.length > 0) { + const { fetchSwarmUserProfile, isSwarmNode } = await import('@/lib/swarm/interactions'); + const { resolveRemoteUser } = await import('@/lib/activitypub/fetch'); + + // Wrap each fetch with a timeout to prevent slow nodes from blocking + const withTimeout = (promise: Promise, ms: number): Promise => { + return Promise.race([ + promise, + new Promise((resolve) => setTimeout(() => resolve(null), ms)) + ]); + }; + + const fetchPromises = followedRemoteUsers.map(async (follow) => { + try { + const atIndex = follow.targetHandle.lastIndexOf('@'); + if (atIndex === -1) return []; + + const handle = follow.targetHandle.slice(0, atIndex); + const domain = follow.targetHandle.slice(atIndex + 1); + + // Check if swarm node - use swarm API (faster) + const isSwarm = await isSwarmNode(domain); + + if (isSwarm) { + const profileData = await withTimeout( + fetchSwarmUserProfile(handle, domain, limit), + 5000 // 5s timeout per node + ); + if (!profileData?.posts) return []; + + return profileData.posts.map(post => ({ + id: `swarm:${domain}:${post.id}`, + content: post.content, + createdAt: new Date(post.createdAt), + likesCount: post.likesCount || 0, + repostsCount: post.repostsCount || 0, + repliesCount: post.repliesCount || 0, + isRemote: true, + isNsfw: post.isNsfw, + linkPreviewUrl: post.linkPreviewUrl, + linkPreviewTitle: post.linkPreviewTitle, + linkPreviewDescription: post.linkPreviewDescription, + linkPreviewImage: post.linkPreviewImage, + author: { + id: `swarm:${domain}:${handle}`, + handle: follow.targetHandle, + displayName: follow.displayName || profileData.profile?.displayName || handle, + avatarUrl: follow.avatarUrl || profileData.profile?.avatarUrl, + isRemote: true, + }, + media: post.media?.map((m: any, idx: number) => ({ + id: `swarm:${domain}:${post.id}:media:${idx}`, + url: m.url, + altText: m.altText || null, + })) || [], + replyTo: null, + })); + } else { + // ActivityPub - fetch from outbox + const remoteProfile = await resolveRemoteUser(handle, domain); + if (!remoteProfile?.outbox) return []; + + // For AP, fall back to cached posts (live outbox fetch is slower) + const cachedPosts = await db.query.remotePosts.findMany({ + where: eq(remotePosts.authorHandle, follow.targetHandle), + orderBy: [desc(remotePosts.publishedAt)], + limit: limit, + }); + + return transformRemotePosts(cachedPosts); + } + } catch (error) { + console.error(`[Home] Error fetching posts from ${follow.targetHandle}:`, error); + return []; + } }); + + const results = await Promise.all(fetchPromises); + liveRemotePosts = results.flat(); } - // Transform remote posts to match local post format (with deduplication) - const transformedRemotePosts = transformRemotePosts(remotePostsData); - // Merge and sort by date - const allPosts = [...localPosts, ...transformedRemotePosts] + const allPosts = [...localPosts, ...liveRemotePosts] .sort((a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime()) .slice(0, limit); diff --git a/src/lib/background/remote-sync.ts b/src/lib/background/remote-sync.ts new file mode 100644 index 0000000..1f4665f --- /dev/null +++ b/src/lib/background/remote-sync.ts @@ -0,0 +1,110 @@ +/** + * Remote Follows Sync + * + * Periodically syncs posts from remote users that local users follow. + * This ensures the home timeline shows fresh posts from followed remote users. + */ + +import { db, remoteFollows } from '@/db'; +import { resolveRemoteUser } from '@/lib/activitypub/fetch'; +import { cacheRemoteUserPosts } from '@/lib/activitypub/cache'; +import { cacheSwarmUserPosts, isSwarmNode } from '@/lib/swarm/interactions'; + +// Track last sync time per remote handle to avoid over-fetching +const lastSyncTimes = new Map(); +const MIN_SYNC_INTERVAL_MS = 60 * 1000; // Don't sync same user more than once per minute + +interface SyncResult { + synced: number; + skipped: number; + errors: number; + details: Array<{ handle: string; cached: number; error?: string }>; +} + +/** + * Sync posts from all remote users that any local user follows + */ +export async function syncRemoteFollowsPosts(origin: string): Promise { + const result: SyncResult = { synced: 0, skipped: 0, errors: 0, details: [] }; + + try { + // Get all unique remote handles that are being followed + const allRemoteFollows = await db.query.remoteFollows.findMany(); + + // Deduplicate by target handle (multiple users might follow the same remote user) + const uniqueHandles = new Map(); + for (const follow of allRemoteFollows) { + if (!uniqueHandles.has(follow.targetHandle)) { + uniqueHandles.set(follow.targetHandle, follow); + } + } + + const now = Date.now(); + + for (const [targetHandle, follow] of uniqueHandles) { + try { + // Check if we've synced this user recently + const lastSync = lastSyncTimes.get(targetHandle); + if (lastSync && (now - lastSync) < MIN_SYNC_INTERVAL_MS) { + result.skipped++; + continue; + } + + // Parse handle to get username and domain + const atIndex = targetHandle.lastIndexOf('@'); + if (atIndex === -1) { + result.skipped++; + continue; + } + + const handle = targetHandle.slice(0, atIndex); + const domain = targetHandle.slice(atIndex + 1); + + // Check if this is a swarm node + const isSwarm = await isSwarmNode(domain); + + let cached = 0; + if (isSwarm) { + // Use swarm sync for swarm nodes + const swarmResult = await cacheSwarmUserPosts(handle, domain, targetHandle, 20); + cached = swarmResult.cached; + } else { + // Use ActivityPub sync for federated nodes + const remoteProfile = await resolveRemoteUser(handle, domain); + if (remoteProfile?.outbox) { + const apResult = await cacheRemoteUserPosts(remoteProfile, targetHandle, origin, 20); + cached = apResult.cached; + } + } + + lastSyncTimes.set(targetHandle, now); + + if (cached > 0) { + result.synced++; + result.details.push({ handle: targetHandle, cached }); + } else { + result.skipped++; + } + } catch (error) { + result.errors++; + result.details.push({ + handle: targetHandle, + cached: 0, + error: error instanceof Error ? error.message : 'Unknown error' + }); + } + } + } catch (error) { + console.error('[RemoteSync] Error syncing remote follows:', error); + result.errors++; + } + + return result; +} + +/** + * Clear the sync cache (useful for testing or forcing a full resync) + */ +export function clearSyncCache(): void { + lastSyncTimes.clear(); +} diff --git a/src/lib/background/scheduler.ts b/src/lib/background/scheduler.ts index e24d889..dc6e2f7 100644 --- a/src/lib/background/scheduler.ts +++ b/src/lib/background/scheduler.ts @@ -4,6 +4,7 @@ * Runs periodic tasks within the Next.js process: * - Bot autonomous posting (every 1 minute) * - Swarm gossip (every 5 minutes) + * - Remote follows sync (every 10 minutes) * - Swarm announcement (on startup) */ @@ -11,9 +12,11 @@ import { processAllAutonomousBots } from '@/lib/bots/autonomous'; import { runGossipRound } from '@/lib/swarm/gossip'; import { announceToSeeds } from '@/lib/swarm/discovery'; import { getSwarmStats } from '@/lib/swarm/registry'; +import { syncRemoteFollowsPosts } from '@/lib/background/remote-sync'; const BOT_INTERVAL_MS = 60 * 1000; // 1 minute const GOSSIP_INTERVAL_MS = 5 * 60 * 1000; // 5 minutes +const REMOTE_SYNC_INTERVAL_MS = 60 * 1000; // 1 minute - keep feeds fresh const STARTUP_DELAY_MS = 10 * 1000; // Wait 10s for server to be ready let isStarted = false; @@ -82,13 +85,33 @@ async function announceToSwarm() { } } -export function startBackgroundTasks() { +async function runRemoteSync(origin: string) { + try { + const result = await syncRemoteFollowsPosts(origin); + if (result.synced > 0 || result.errors > 0) { + log('REMOTE_SYNC', `Synced ${result.synced} users, skipped ${result.skipped}, errors ${result.errors}`); + if (result.details.length > 0) { + const newPosts = result.details.filter(d => d.cached > 0); + if (newPosts.length > 0) { + log('REMOTE_SYNC', `New posts: ${newPosts.map(d => `${d.handle}: ${d.cached}`).join(', ')}`); + } + } + } + } catch (error) { + log('REMOTE_SYNC', `Error: ${error}`); + } +} + +export function startBackgroundTasks(origin?: string) { // Prevent double-start (Next.js can call register() multiple times in dev) if (isStarted) return; isStarted = true; + // Default origin for remote sync (can be overridden) + const syncOrigin = origin || process.env.NEXT_PUBLIC_APP_URL || 'http://localhost:3000'; + log('STARTUP', 'Background task scheduler starting...'); - log('STARTUP', `Bot interval: ${BOT_INTERVAL_MS / 1000}s, Gossip interval: ${GOSSIP_INTERVAL_MS / 1000}s`); + log('STARTUP', `Bot interval: ${BOT_INTERVAL_MS / 1000}s, Gossip interval: ${GOSSIP_INTERVAL_MS / 1000}s, Remote sync interval: ${REMOTE_SYNC_INTERVAL_MS / 1000}s`); // Wait for server to be fully ready before starting tasks setTimeout(async () => { @@ -100,9 +123,13 @@ export function startBackgroundTasks() { // Run initial bot check await runBotTasks(); + // Run initial remote sync (after 15s to let server stabilize) + setTimeout(() => runRemoteSync(syncOrigin), 15 * 1000); + // Schedule recurring tasks setInterval(runBotTasks, BOT_INTERVAL_MS); setInterval(runSwarmGossip, GOSSIP_INTERVAL_MS); + setInterval(() => runRemoteSync(syncOrigin), REMOTE_SYNC_INTERVAL_MS); // First gossip after 30s (let announcement propagate) setTimeout(runSwarmGossip, 30 * 1000); diff --git a/src/lib/bots/autonomous.ts b/src/lib/bots/autonomous.ts index c5ca18f..33fb430 100644 --- a/src/lib/bots/autonomous.ts +++ b/src/lib/bots/autonomous.ts @@ -11,9 +11,10 @@ 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 { canPost } from './rateLimiter'; import { decryptApiKey, deserializeEncryptedData } from './encryption'; import { parseScheduleConfig, isDue } from './scheduler'; +import { triggerPost } from './posting'; // ============================================ // TYPES @@ -446,9 +447,6 @@ export async function attemptAutonomousPost( ); } - const contentGeneratorBot = toContentGeneratorBot(dbBot, dbBot.user.handle); - const generator = new ContentGenerator(contentGeneratorBot); - // Check if bot has content sources const hasSourcesConfigured = await hasContentSources(botId); @@ -471,25 +469,25 @@ export async function attemptAutonomousPost( ); if (evaluation.shouldPost) { - const generatedContent = await generator.generatePost(contentItem); + // Use triggerPost to actually create the post in the database + const postResult = await triggerPost(botId, { + sourceContentId: contentItem.id, + skipRateLimitCheck, // Already checked above + }); - // Record the post for rate limiting - if (!skipRateLimitCheck) { - await recordPost(botId); + if (postResult.success) { + return { + posted: true, + postId: postResult.post?.id, + postText: postResult.post?.content, + contentItem, + }; + } else { + throw new AutonomousPostError( + postResult.error || 'Failed to create post', + postResult.errorCode as AutonomousPostErrorCode || 'POST_CREATION_FAILED' + ); } - - await markContentItemProcessed( - contentItem.id, - 'pending', - evaluation.interestScore, - evaluation.reason - ); - - return { - posted: true, - postText: generatedContent.text, - contentItem, - }; } } catch (error) { console.error(`Error evaluating content item ${contentItem.id}:`, error); @@ -507,24 +505,21 @@ export async function attemptAutonomousPost( } // 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); - } + // Use triggerPost which handles database insertion, rate limiting, and federation + const postResult = await triggerPost(botId, { + skipRateLimitCheck, // Already checked above + }); + if (postResult.success) { return { posted: true, - postText: generatedContent.text, + postId: postResult.post?.id, + postText: postResult.post?.content, }; - } catch (error) { - if (error instanceof AutonomousPostError) throw error; + } else { throw new AutonomousPostError( - `Failed to create post: ${error instanceof Error ? error.message : String(error)}`, - 'POST_CREATION_FAILED', - error instanceof Error ? error : undefined + postResult.error || 'Failed to create post', + postResult.errorCode as AutonomousPostErrorCode || 'POST_CREATION_FAILED' ); } }