feat(timeline,background): Add live remote post fetching and background sync

- Implement live post fetching from followed remote users with parallel requests and timeouts
- Add remote-sync background task to periodically cache posts from followed remote users
- Optimize swarm node detection and prioritize swarm API over ActivityPub for faster responses
- Add timeout wrapper (5s per node) to prevent slow remote nodes from blocking timeline requests
- Enhance scheduler to support background sync tasks with configurable intervals
- Transform remote posts to match local format with proper media and author metadata
- Improve error handling and logging for remote user post fetching
- Replace cached-only approach with hybrid live/cached strategy for fresher content
This commit is contained in:
Christomatt
2026-01-26 10:14:07 +01:00
parent e98221e81a
commit 728dda6b52
4 changed files with 249 additions and 48 deletions
+79 -10
View File
@@ -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),
// 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 = <T>(promise: Promise<T>, ms: number): Promise<T | null> => {
return Promise.race([
promise,
new Promise<null>((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);
+110
View File
@@ -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<string, number>();
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<SyncResult> {
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<string, typeof allRemoteFollows[0]>();
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();
}
+29 -2
View File
@@ -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);
+26 -31
View File
@@ -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);
// Record the post for rate limiting
if (!skipRateLimitCheck) {
await recordPost(botId);
}
await markContentItemProcessed(
contentItem.id,
'pending',
evaluation.interestScore,
evaluation.reason
);
// Use triggerPost to actually create the post in the database
const postResult = await triggerPost(botId, {
sourceContentId: contentItem.id,
skipRateLimitCheck, // Already checked above
});
if (postResult.success) {
return {
posted: true,
postText: generatedContent.text,
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'
);
}
}
} 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'
);
}
}