feat(swarm): Implement decentralized node discovery and content federation system
- Add swarm node registry with discovery, gossip, and timeline synchronization - Implement database schema for swarm nodes, seeds, and sync logs with proper indexing - Create swarm API endpoints for node discovery, gossip protocol, timeline sharing, and announcements - Add content moderation features with NSFW flagging and muted nodes support - Implement user settings pages for content filtering and node moderation - Add background scheduler for automated swarm synchronization and node health checks - Create .well-known endpoint for swarm protocol discovery - Remove legacy bot-cron.ts in favor of integrated scheduler system - Add user account NSFW preferences and age verification tracking - Update database schema with swarm-related tables and user moderation fields - Enhance post and user components to support federated content display - Add instrumentation for monitoring swarm operations and node health
This commit is contained in:
AskIt
@@ -0,0 +1,200 @@
|
||||
/**
|
||||
* Swarm Discovery
|
||||
*
|
||||
* Handles node discovery and announcement in the swarm network.
|
||||
*/
|
||||
|
||||
import { db, nodes, users, posts } from '@/db';
|
||||
import { eq, sql } from 'drizzle-orm';
|
||||
import type { SwarmAnnouncement, SwarmNodeInfo, SwarmCapability } from './types';
|
||||
import { upsertSwarmNode, getSeedNodes, markNodeSuccess, markNodeFailure } from './registry';
|
||||
|
||||
/**
|
||||
* Build this node's announcement payload
|
||||
*/
|
||||
export async function buildAnnouncement(): Promise<SwarmAnnouncement> {
|
||||
const domain = process.env.NEXT_PUBLIC_NODE_DOMAIN || 'localhost:3000';
|
||||
|
||||
let name = 'Synapsis Node';
|
||||
let description: string | undefined;
|
||||
let logoUrl: string | undefined;
|
||||
let publicKey = '';
|
||||
let userCount = 0;
|
||||
let postCount = 0;
|
||||
let isNsfw = false;
|
||||
|
||||
if (db) {
|
||||
// Get node info
|
||||
const node = await db.query.nodes.findFirst({
|
||||
where: eq(nodes.domain, domain),
|
||||
});
|
||||
|
||||
if (node) {
|
||||
name = node.name;
|
||||
description = node.description ?? undefined;
|
||||
logoUrl = node.logoUrl ?? undefined;
|
||||
publicKey = node.publicKey ?? '';
|
||||
isNsfw = node.isNsfw;
|
||||
}
|
||||
|
||||
// Get counts
|
||||
const userResult = await db.select({ count: sql<number>`count(*)` }).from(users);
|
||||
const postResult = await db.select({ count: sql<number>`count(*)` }).from(posts);
|
||||
|
||||
userCount = Number(userResult[0]?.count ?? 0);
|
||||
postCount = Number(postResult[0]?.count ?? 0);
|
||||
}
|
||||
|
||||
const capabilities: SwarmCapability[] = ['handles', 'gossip'];
|
||||
|
||||
return {
|
||||
domain,
|
||||
name,
|
||||
description,
|
||||
logoUrl,
|
||||
publicKey,
|
||||
softwareVersion: '0.1.0', // TODO: Get from package.json
|
||||
userCount,
|
||||
postCount,
|
||||
capabilities,
|
||||
isNsfw,
|
||||
timestamp: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Announce this node to a remote node
|
||||
*/
|
||||
export async function announceToNode(targetDomain: string): Promise<{ success: boolean; error?: string }> {
|
||||
try {
|
||||
const announcement = await buildAnnouncement();
|
||||
|
||||
const baseUrl = targetDomain.startsWith('http') ? targetDomain : `https://${targetDomain}`;
|
||||
const url = `${baseUrl}/api/swarm/announce`;
|
||||
|
||||
const response = await fetch(url, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'Accept': 'application/json',
|
||||
},
|
||||
body: JSON.stringify(announcement),
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
const error = await response.text();
|
||||
await markNodeFailure(targetDomain);
|
||||
return { success: false, error: `HTTP ${response.status}: ${error}` };
|
||||
}
|
||||
|
||||
// The remote node should respond with their info
|
||||
const remoteInfo = await response.json() as SwarmNodeInfo;
|
||||
|
||||
// Add/update the remote node in our registry
|
||||
await upsertSwarmNode(remoteInfo, 'direct');
|
||||
await markNodeSuccess(targetDomain);
|
||||
|
||||
return { success: true };
|
||||
} catch (error) {
|
||||
await markNodeFailure(targetDomain);
|
||||
return {
|
||||
success: false,
|
||||
error: error instanceof Error ? error.message : 'Unknown error'
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Announce to all seed nodes (bootstrap)
|
||||
*/
|
||||
export async function announceToSeeds(): Promise<{
|
||||
successful: string[];
|
||||
failed: { domain: string; error: string }[]
|
||||
}> {
|
||||
const seeds = await getSeedNodes();
|
||||
const ourDomain = process.env.NEXT_PUBLIC_NODE_DOMAIN;
|
||||
|
||||
// Don't announce to ourselves
|
||||
const targetSeeds = seeds.filter(s => s !== ourDomain);
|
||||
|
||||
const successful: string[] = [];
|
||||
const failed: { domain: string; error: string }[] = [];
|
||||
|
||||
for (const seed of targetSeeds) {
|
||||
const result = await announceToNode(seed);
|
||||
if (result.success) {
|
||||
successful.push(seed);
|
||||
} else {
|
||||
failed.push({ domain: seed, error: result.error || 'Unknown error' });
|
||||
}
|
||||
}
|
||||
|
||||
return { successful, failed };
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch node info from a remote node
|
||||
*/
|
||||
export async function fetchNodeInfo(domain: string): Promise<SwarmNodeInfo | null> {
|
||||
try {
|
||||
const baseUrl = domain.startsWith('http') ? domain : `https://${domain}`;
|
||||
|
||||
// Try the swarm endpoint first
|
||||
let response = await fetch(`${baseUrl}/api/swarm/info`, {
|
||||
headers: { 'Accept': 'application/json' },
|
||||
});
|
||||
|
||||
if (!response.ok) {
|
||||
// Fall back to standard node endpoint
|
||||
response = await fetch(`${baseUrl}/api/node`, {
|
||||
headers: { 'Accept': 'application/json' },
|
||||
});
|
||||
}
|
||||
|
||||
if (!response.ok) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const data = await response.json();
|
||||
|
||||
return {
|
||||
domain: data.domain || domain,
|
||||
name: data.name,
|
||||
description: data.description,
|
||||
logoUrl: data.logoUrl,
|
||||
publicKey: data.publicKey,
|
||||
softwareVersion: data.softwareVersion,
|
||||
userCount: data.userCount,
|
||||
postCount: data.postCount,
|
||||
capabilities: data.capabilities,
|
||||
isNsfw: data.isNsfw,
|
||||
};
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Discover a node and add it to the registry
|
||||
*/
|
||||
export async function discoverNode(
|
||||
domain: string,
|
||||
discoveredVia?: string
|
||||
): Promise<{ success: boolean; isNew: boolean; error?: string }> {
|
||||
const ourDomain = process.env.NEXT_PUBLIC_NODE_DOMAIN;
|
||||
|
||||
// Don't discover ourselves
|
||||
if (domain === ourDomain) {
|
||||
return { success: false, isNew: false, error: 'Cannot discover self' };
|
||||
}
|
||||
|
||||
const info = await fetchNodeInfo(domain);
|
||||
|
||||
if (!info) {
|
||||
return { success: false, isNew: false, error: 'Could not fetch node info' };
|
||||
}
|
||||
|
||||
const result = await upsertSwarmNode(info, discoveredVia);
|
||||
|
||||
return { success: true, isNew: result.isNew };
|
||||
}
|
||||
@@ -0,0 +1,237 @@
|
||||
/**
|
||||
* Swarm Gossip Protocol
|
||||
*
|
||||
* Implements epidemic-style gossip for node and handle propagation.
|
||||
* Nodes periodically exchange their known nodes/handles with random peers.
|
||||
*/
|
||||
|
||||
import { db, handleRegistry } from '@/db';
|
||||
import { desc, gt } from 'drizzle-orm';
|
||||
import type { SwarmGossipPayload, SwarmGossipResponse, SwarmSyncResult, SwarmNodeInfo } from './types';
|
||||
import { SWARM_CONFIG } from './types';
|
||||
import {
|
||||
getNodesForGossip,
|
||||
getActiveSwarmNodes,
|
||||
getNodesSince,
|
||||
upsertSwarmNodes,
|
||||
markNodeSuccess,
|
||||
markNodeFailure,
|
||||
logSync,
|
||||
} from './registry';
|
||||
import { upsertHandleEntries } from '@/lib/federation/handles';
|
||||
import { buildAnnouncement } from './discovery';
|
||||
|
||||
/**
|
||||
* Build a gossip payload to send to another node
|
||||
*/
|
||||
export async function buildGossipPayload(since?: string): Promise<SwarmGossipPayload> {
|
||||
const ourDomain = process.env.NEXT_PUBLIC_NODE_DOMAIN || 'localhost:3000';
|
||||
|
||||
// Get nodes to share
|
||||
let nodes: SwarmNodeInfo[];
|
||||
if (since) {
|
||||
nodes = await getNodesSince(new Date(since), SWARM_CONFIG.maxNodesPerGossip);
|
||||
} else {
|
||||
nodes = await getActiveSwarmNodes(SWARM_CONFIG.maxNodesPerGossip);
|
||||
}
|
||||
|
||||
// Include ourselves in the node list
|
||||
const announcement = await buildAnnouncement();
|
||||
const selfNode: SwarmNodeInfo = {
|
||||
domain: announcement.domain,
|
||||
name: announcement.name,
|
||||
description: announcement.description,
|
||||
logoUrl: announcement.logoUrl,
|
||||
publicKey: announcement.publicKey,
|
||||
softwareVersion: announcement.softwareVersion,
|
||||
userCount: announcement.userCount,
|
||||
postCount: announcement.postCount,
|
||||
capabilities: announcement.capabilities,
|
||||
lastSeenAt: new Date().toISOString(),
|
||||
};
|
||||
|
||||
// Get handles to share
|
||||
let handles: SwarmGossipPayload['handles'] = [];
|
||||
if (db) {
|
||||
const sinceDate = since ? new Date(since) : undefined;
|
||||
const handleEntries = await db.query.handleRegistry.findMany({
|
||||
where: sinceDate ? gt(handleRegistry.updatedAt, sinceDate) : undefined,
|
||||
orderBy: [desc(handleRegistry.updatedAt)],
|
||||
limit: SWARM_CONFIG.maxHandlesPerGossip,
|
||||
});
|
||||
|
||||
handles = handleEntries.map(h => ({
|
||||
handle: h.handle,
|
||||
did: h.did,
|
||||
nodeDomain: h.nodeDomain,
|
||||
updatedAt: h.updatedAt?.toISOString(),
|
||||
}));
|
||||
}
|
||||
|
||||
return {
|
||||
sender: ourDomain,
|
||||
nodes: [selfNode, ...nodes],
|
||||
handles,
|
||||
timestamp: new Date().toISOString(),
|
||||
since,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Process incoming gossip and return our response
|
||||
*/
|
||||
export async function processGossip(
|
||||
payload: SwarmGossipPayload
|
||||
): Promise<SwarmGossipResponse> {
|
||||
const startTime = Date.now();
|
||||
|
||||
// Process incoming nodes
|
||||
const nodeResult = await upsertSwarmNodes(payload.nodes, payload.sender);
|
||||
|
||||
// Process incoming handles
|
||||
let handlesResult = { added: 0, updated: 0 };
|
||||
if (payload.handles && payload.handles.length > 0) {
|
||||
handlesResult = await upsertHandleEntries(payload.handles);
|
||||
}
|
||||
|
||||
// Build our response with nodes/handles to share back
|
||||
const responsePayload = await buildGossipPayload(payload.since);
|
||||
|
||||
return {
|
||||
nodes: responsePayload.nodes,
|
||||
handles: responsePayload.handles,
|
||||
received: {
|
||||
nodes: nodeResult.added + nodeResult.updated,
|
||||
handles: handlesResult.added + handlesResult.updated,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Send gossip to a specific node
|
||||
*/
|
||||
export async function gossipToNode(
|
||||
targetDomain: string,
|
||||
since?: string
|
||||
): Promise<SwarmSyncResult> {
|
||||
const startTime = Date.now();
|
||||
|
||||
try {
|
||||
const payload = await buildGossipPayload(since);
|
||||
|
||||
const baseUrl = targetDomain.startsWith('http') ? targetDomain : `https://${targetDomain}`;
|
||||
const url = `${baseUrl}/api/swarm/gossip`;
|
||||
|
||||
const response = await fetch(url, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'Accept': 'application/json',
|
||||
},
|
||||
body: JSON.stringify(payload),
|
||||
});
|
||||
|
||||
const durationMs = Date.now() - startTime;
|
||||
|
||||
if (!response.ok) {
|
||||
const error = `HTTP ${response.status}`;
|
||||
await markNodeFailure(targetDomain);
|
||||
await logSync(targetDomain, 'push', {
|
||||
success: false,
|
||||
nodesReceived: 0,
|
||||
nodesSent: payload.nodes.length,
|
||||
handlesReceived: 0,
|
||||
handlesSent: payload.handles?.length || 0,
|
||||
error,
|
||||
durationMs,
|
||||
});
|
||||
return {
|
||||
success: false,
|
||||
nodesReceived: 0,
|
||||
nodesSent: payload.nodes.length,
|
||||
handlesReceived: 0,
|
||||
handlesSent: payload.handles?.length || 0,
|
||||
error,
|
||||
durationMs,
|
||||
};
|
||||
}
|
||||
|
||||
const gossipResponse = await response.json() as SwarmGossipResponse;
|
||||
|
||||
// Process the response (nodes and handles they sent back)
|
||||
const nodeResult = await upsertSwarmNodes(gossipResponse.nodes, targetDomain);
|
||||
|
||||
let handlesResult = { added: 0, updated: 0 };
|
||||
if (gossipResponse.handles && gossipResponse.handles.length > 0) {
|
||||
handlesResult = await upsertHandleEntries(gossipResponse.handles);
|
||||
}
|
||||
|
||||
await markNodeSuccess(targetDomain);
|
||||
|
||||
const result: SwarmSyncResult = {
|
||||
success: true,
|
||||
nodesReceived: nodeResult.added + nodeResult.updated,
|
||||
nodesSent: payload.nodes.length,
|
||||
handlesReceived: handlesResult.added + handlesResult.updated,
|
||||
handlesSent: payload.handles?.length || 0,
|
||||
durationMs,
|
||||
};
|
||||
|
||||
await logSync(targetDomain, 'push', result);
|
||||
return result;
|
||||
} catch (error) {
|
||||
const durationMs = Date.now() - startTime;
|
||||
const errorMsg = error instanceof Error ? error.message : 'Unknown error';
|
||||
|
||||
await markNodeFailure(targetDomain);
|
||||
|
||||
const result: SwarmSyncResult = {
|
||||
success: false,
|
||||
nodesReceived: 0,
|
||||
nodesSent: 0,
|
||||
handlesReceived: 0,
|
||||
handlesSent: 0,
|
||||
error: errorMsg,
|
||||
durationMs,
|
||||
};
|
||||
|
||||
await logSync(targetDomain, 'push', result);
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Run a gossip round - contact random nodes and exchange info
|
||||
*/
|
||||
export async function runGossipRound(): Promise<{
|
||||
contacted: number;
|
||||
successful: number;
|
||||
totalNodesReceived: number;
|
||||
totalHandlesReceived: number;
|
||||
}> {
|
||||
// Get random nodes to gossip with
|
||||
const targets = await getNodesForGossip(SWARM_CONFIG.gossipFanout);
|
||||
|
||||
let contacted = 0;
|
||||
let successful = 0;
|
||||
let totalNodesReceived = 0;
|
||||
let totalHandlesReceived = 0;
|
||||
|
||||
for (const target of targets) {
|
||||
contacted++;
|
||||
const result = await gossipToNode(target.domain);
|
||||
|
||||
if (result.success) {
|
||||
successful++;
|
||||
totalNodesReceived += result.nodesReceived;
|
||||
totalHandlesReceived += result.handlesReceived;
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
contacted,
|
||||
successful,
|
||||
totalNodesReceived,
|
||||
totalHandlesReceived,
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
/**
|
||||
* Swarm Module
|
||||
*
|
||||
* The Synapsis swarm is a decentralized network of nodes that discover
|
||||
* and communicate with each other using a hybrid approach:
|
||||
*
|
||||
* 1. Seed nodes (like node.synapsis.social) provide initial bootstrap
|
||||
* 2. Gossip protocol spreads node/handle information epidemically
|
||||
* 3. Any node can discover the full network without central authority
|
||||
*
|
||||
* Usage:
|
||||
* - On node startup: call announceToSeeds() to register with the network
|
||||
* - Periodically: call runGossipRound() to exchange info with peers
|
||||
* - On demand: call discoverNode() to add a specific node
|
||||
*/
|
||||
|
||||
export * from './types';
|
||||
export * from './registry';
|
||||
export * from './discovery';
|
||||
export * from './gossip';
|
||||
@@ -0,0 +1,311 @@
|
||||
/**
|
||||
* Swarm Registry
|
||||
*
|
||||
* Manages the local registry of known swarm nodes.
|
||||
*/
|
||||
|
||||
import { db, swarmNodes, swarmSeeds, swarmSyncLog } from '@/db';
|
||||
import { eq, desc, and, gt, lt, sql } from 'drizzle-orm';
|
||||
import type { SwarmNodeInfo, SwarmCapability, SwarmSyncResult } from './types';
|
||||
import { SWARM_CONFIG, DEFAULT_SEED_NODES } from './types';
|
||||
|
||||
/**
|
||||
* Get or create a swarm node entry
|
||||
*/
|
||||
export async function upsertSwarmNode(
|
||||
node: SwarmNodeInfo,
|
||||
discoveredVia?: string
|
||||
): Promise<{ isNew: boolean }> {
|
||||
if (!db) {
|
||||
return { isNew: false };
|
||||
}
|
||||
|
||||
const existing = await db.query.swarmNodes.findFirst({
|
||||
where: eq(swarmNodes.domain, node.domain),
|
||||
});
|
||||
|
||||
const capabilities = node.capabilities ? JSON.stringify(node.capabilities) : null;
|
||||
|
||||
if (!existing) {
|
||||
await db.insert(swarmNodes).values({
|
||||
domain: node.domain,
|
||||
name: node.name,
|
||||
description: node.description,
|
||||
logoUrl: node.logoUrl,
|
||||
publicKey: node.publicKey,
|
||||
softwareVersion: node.softwareVersion,
|
||||
userCount: node.userCount,
|
||||
postCount: node.postCount,
|
||||
isNsfw: node.isNsfw ?? false,
|
||||
discoveredVia,
|
||||
capabilities,
|
||||
lastSeenAt: node.lastSeenAt ? new Date(node.lastSeenAt) : new Date(),
|
||||
});
|
||||
return { isNew: true };
|
||||
}
|
||||
|
||||
// Update existing node
|
||||
await db.update(swarmNodes)
|
||||
.set({
|
||||
name: node.name ?? existing.name,
|
||||
description: node.description ?? existing.description,
|
||||
logoUrl: node.logoUrl ?? existing.logoUrl,
|
||||
publicKey: node.publicKey ?? existing.publicKey,
|
||||
softwareVersion: node.softwareVersion ?? existing.softwareVersion,
|
||||
userCount: node.userCount ?? existing.userCount,
|
||||
postCount: node.postCount ?? existing.postCount,
|
||||
isNsfw: node.isNsfw ?? existing.isNsfw,
|
||||
capabilities: capabilities ?? existing.capabilities,
|
||||
lastSeenAt: new Date(),
|
||||
consecutiveFailures: 0,
|
||||
isActive: true,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(swarmNodes.domain, node.domain));
|
||||
|
||||
return { isNew: false };
|
||||
}
|
||||
|
||||
/**
|
||||
* Bulk upsert swarm nodes from gossip
|
||||
*/
|
||||
export async function upsertSwarmNodes(
|
||||
nodes: SwarmNodeInfo[],
|
||||
discoveredVia: string
|
||||
): Promise<{ added: number; updated: number }> {
|
||||
if (!db || nodes.length === 0) {
|
||||
return { added: 0, updated: 0 };
|
||||
}
|
||||
|
||||
let added = 0;
|
||||
let updated = 0;
|
||||
|
||||
// Filter out our own domain
|
||||
const ourDomain = process.env.NEXT_PUBLIC_NODE_DOMAIN;
|
||||
const filteredNodes = nodes.filter(n => n.domain !== ourDomain);
|
||||
|
||||
for (const node of filteredNodes) {
|
||||
const result = await upsertSwarmNode(node, discoveredVia);
|
||||
if (result.isNew) {
|
||||
added++;
|
||||
} else {
|
||||
updated++;
|
||||
}
|
||||
}
|
||||
|
||||
return { added, updated };
|
||||
}
|
||||
|
||||
/**
|
||||
* Get all active swarm nodes
|
||||
*/
|
||||
export async function getActiveSwarmNodes(limit = 100): Promise<SwarmNodeInfo[]> {
|
||||
if (!db) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const nodes = await db.query.swarmNodes.findMany({
|
||||
where: eq(swarmNodes.isActive, true),
|
||||
orderBy: [desc(swarmNodes.lastSeenAt)],
|
||||
limit,
|
||||
});
|
||||
|
||||
return nodes.map(nodeToInfo);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get nodes for gossip (random selection of active nodes)
|
||||
*/
|
||||
export async function getNodesForGossip(count: number): Promise<SwarmNodeInfo[]> {
|
||||
if (!db) {
|
||||
return [];
|
||||
}
|
||||
|
||||
// Get active nodes with decent trust scores, ordered randomly
|
||||
const nodes = await db.query.swarmNodes.findMany({
|
||||
where: and(
|
||||
eq(swarmNodes.isActive, true),
|
||||
gt(swarmNodes.trustScore, 20)
|
||||
),
|
||||
orderBy: sql`RANDOM()`,
|
||||
limit: count,
|
||||
});
|
||||
|
||||
return nodes.map(nodeToInfo);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get nodes updated since a timestamp (for incremental sync)
|
||||
*/
|
||||
export async function getNodesSince(since: Date, limit = 100): Promise<SwarmNodeInfo[]> {
|
||||
if (!db) {
|
||||
return [];
|
||||
}
|
||||
|
||||
const nodes = await db.query.swarmNodes.findMany({
|
||||
where: gt(swarmNodes.updatedAt, since),
|
||||
orderBy: [desc(swarmNodes.updatedAt)],
|
||||
limit,
|
||||
});
|
||||
|
||||
return nodes.map(nodeToInfo);
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark a node as having failed contact
|
||||
*/
|
||||
export async function markNodeFailure(domain: string): Promise<void> {
|
||||
if (!db) return;
|
||||
|
||||
const node = await db.query.swarmNodes.findFirst({
|
||||
where: eq(swarmNodes.domain, domain),
|
||||
});
|
||||
|
||||
if (!node) return;
|
||||
|
||||
const newFailures = node.consecutiveFailures + 1;
|
||||
const newTrust = Math.max(
|
||||
SWARM_CONFIG.minTrustScore,
|
||||
node.trustScore + SWARM_CONFIG.trustScoreOnFailure
|
||||
);
|
||||
const isActive = newFailures < SWARM_CONFIG.maxConsecutiveFailures;
|
||||
|
||||
await db.update(swarmNodes)
|
||||
.set({
|
||||
consecutiveFailures: newFailures,
|
||||
trustScore: newTrust,
|
||||
isActive,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(swarmNodes.domain, domain));
|
||||
}
|
||||
|
||||
/**
|
||||
* Mark a node as successfully contacted
|
||||
*/
|
||||
export async function markNodeSuccess(domain: string): Promise<void> {
|
||||
if (!db) return;
|
||||
|
||||
const node = await db.query.swarmNodes.findFirst({
|
||||
where: eq(swarmNodes.domain, domain),
|
||||
});
|
||||
|
||||
if (!node) return;
|
||||
|
||||
const newTrust = Math.min(
|
||||
SWARM_CONFIG.maxTrustScore,
|
||||
node.trustScore + SWARM_CONFIG.trustScoreOnSuccess
|
||||
);
|
||||
|
||||
await db.update(swarmNodes)
|
||||
.set({
|
||||
consecutiveFailures: 0,
|
||||
trustScore: newTrust,
|
||||
isActive: true,
|
||||
lastSeenAt: new Date(),
|
||||
lastSyncAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(swarmNodes.domain, domain));
|
||||
}
|
||||
|
||||
/**
|
||||
* Log a sync operation
|
||||
*/
|
||||
export async function logSync(
|
||||
remoteDomain: string,
|
||||
direction: 'push' | 'pull',
|
||||
result: SwarmSyncResult
|
||||
): Promise<void> {
|
||||
if (!db) return;
|
||||
|
||||
await db.insert(swarmSyncLog).values({
|
||||
remoteDomain,
|
||||
direction,
|
||||
nodesReceived: result.nodesReceived,
|
||||
nodesSent: result.nodesSent,
|
||||
handlesReceived: result.handlesReceived,
|
||||
handlesSent: result.handlesSent,
|
||||
success: result.success,
|
||||
errorMessage: result.error,
|
||||
durationMs: result.durationMs,
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Get seed nodes (with fallback to defaults)
|
||||
*/
|
||||
export async function getSeedNodes(): Promise<string[]> {
|
||||
if (!db) {
|
||||
return [...DEFAULT_SEED_NODES];
|
||||
}
|
||||
|
||||
const seeds = await db.query.swarmSeeds.findMany({
|
||||
where: eq(swarmSeeds.isEnabled, true),
|
||||
orderBy: [swarmSeeds.priority],
|
||||
});
|
||||
|
||||
if (seeds.length === 0) {
|
||||
return [...DEFAULT_SEED_NODES];
|
||||
}
|
||||
|
||||
return seeds.map(s => s.domain);
|
||||
}
|
||||
|
||||
/**
|
||||
* Add a seed node
|
||||
*/
|
||||
export async function addSeedNode(domain: string, priority = 100): Promise<void> {
|
||||
if (!db) return;
|
||||
|
||||
await db.insert(swarmSeeds)
|
||||
.values({ domain, priority })
|
||||
.onConflictDoUpdate({
|
||||
target: swarmSeeds.domain,
|
||||
set: { priority, isEnabled: true },
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* Get swarm statistics
|
||||
*/
|
||||
export async function getSwarmStats() {
|
||||
if (!db) {
|
||||
return {
|
||||
totalNodes: 0,
|
||||
activeNodes: 0,
|
||||
totalUsers: 0,
|
||||
totalPosts: 0,
|
||||
};
|
||||
}
|
||||
|
||||
const allNodes = await db.query.swarmNodes.findMany();
|
||||
const activeNodes = allNodes.filter(n => n.isActive);
|
||||
|
||||
const totalUsers = activeNodes.reduce((sum, n) => sum + (n.userCount || 0), 0);
|
||||
const totalPosts = activeNodes.reduce((sum, n) => sum + (n.postCount || 0), 0);
|
||||
|
||||
return {
|
||||
totalNodes: allNodes.length,
|
||||
activeNodes: activeNodes.length,
|
||||
totalUsers,
|
||||
totalPosts,
|
||||
};
|
||||
}
|
||||
|
||||
// Helper to convert DB node to SwarmNodeInfo
|
||||
function nodeToInfo(node: typeof swarmNodes.$inferSelect): SwarmNodeInfo {
|
||||
return {
|
||||
domain: node.domain,
|
||||
name: node.name ?? undefined,
|
||||
description: node.description ?? undefined,
|
||||
logoUrl: node.logoUrl ?? undefined,
|
||||
publicKey: node.publicKey ?? undefined,
|
||||
softwareVersion: node.softwareVersion ?? undefined,
|
||||
userCount: node.userCount ?? undefined,
|
||||
postCount: node.postCount ?? undefined,
|
||||
capabilities: node.capabilities ? JSON.parse(node.capabilities) : undefined,
|
||||
isNsfw: node.isNsfw,
|
||||
lastSeenAt: node.lastSeenAt.toISOString(),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,129 @@
|
||||
/**
|
||||
* Swarm Timeline
|
||||
*
|
||||
* Fetches and aggregates posts from across the swarm network.
|
||||
*/
|
||||
|
||||
import { getActiveSwarmNodes } from './registry';
|
||||
import type { SwarmPost } from '@/app/api/swarm/timeline/route';
|
||||
|
||||
interface TimelineResult {
|
||||
posts: SwarmPost[];
|
||||
sources: { domain: string; postCount: number; isNsfw?: boolean; error?: string }[];
|
||||
fetchedAt: string;
|
||||
}
|
||||
|
||||
interface TimelineOptions {
|
||||
includeNsfw?: boolean; // Whether to include NSFW content
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch timeline from a single node
|
||||
*/
|
||||
async function fetchNodeTimeline(
|
||||
domain: string,
|
||||
limit: number = 20
|
||||
): Promise<{ posts: SwarmPost[]; nodeIsNsfw?: boolean; error?: string }> {
|
||||
try {
|
||||
const baseUrl = domain.startsWith('http') ? domain : `https://${domain}`;
|
||||
const url = `${baseUrl}/api/swarm/timeline?limit=${limit}`;
|
||||
|
||||
const controller = new AbortController();
|
||||
const timeout = setTimeout(() => controller.abort(), 5000); // 5s timeout
|
||||
|
||||
const response = await fetch(url, {
|
||||
headers: { 'Accept': 'application/json' },
|
||||
signal: controller.signal,
|
||||
});
|
||||
|
||||
clearTimeout(timeout);
|
||||
|
||||
if (!response.ok) {
|
||||
return { posts: [], error: `HTTP ${response.status}` };
|
||||
}
|
||||
|
||||
const data = await response.json();
|
||||
return { posts: data.posts || [], nodeIsNsfw: data.nodeIsNsfw };
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : 'Unknown error';
|
||||
return { posts: [], error: message };
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Fetch aggregated timeline from the swarm
|
||||
*
|
||||
* Queries multiple nodes in parallel and merges results.
|
||||
* Filters out NSFW content unless explicitly requested.
|
||||
*/
|
||||
export async function fetchSwarmTimeline(
|
||||
maxNodes: number = 10,
|
||||
postsPerNode: number = 10,
|
||||
options: TimelineOptions = {}
|
||||
): Promise<TimelineResult> {
|
||||
const { includeNsfw = false } = options;
|
||||
|
||||
// Get active nodes to query
|
||||
const nodes = await getActiveSwarmNodes(maxNodes);
|
||||
|
||||
// Always include our own posts
|
||||
const ourDomain = process.env.NEXT_PUBLIC_NODE_DOMAIN || 'localhost';
|
||||
|
||||
// Filter out NSFW nodes if not including NSFW content
|
||||
const eligibleNodes = includeNsfw
|
||||
? nodes
|
||||
: nodes.filter(n => !n.isNsfw);
|
||||
|
||||
const nodesToQuery = [
|
||||
ourDomain,
|
||||
...eligibleNodes.map(n => n.domain).filter(d => d !== ourDomain)
|
||||
].slice(0, maxNodes);
|
||||
|
||||
// Fetch from all nodes in parallel
|
||||
const results = await Promise.all(
|
||||
nodesToQuery.map(async (domain) => {
|
||||
const result = await fetchNodeTimeline(domain, postsPerNode);
|
||||
return {
|
||||
domain,
|
||||
...result,
|
||||
};
|
||||
})
|
||||
);
|
||||
|
||||
// Collect all posts and track sources
|
||||
const allPosts: SwarmPost[] = [];
|
||||
const sources: TimelineResult['sources'] = [];
|
||||
|
||||
for (const result of results) {
|
||||
sources.push({
|
||||
domain: result.domain,
|
||||
postCount: result.posts.length,
|
||||
isNsfw: result.nodeIsNsfw,
|
||||
error: result.error,
|
||||
});
|
||||
|
||||
// Filter NSFW posts if not including NSFW
|
||||
const filteredPosts = includeNsfw
|
||||
? result.posts
|
||||
: result.posts.filter(p => !p.isNsfw);
|
||||
|
||||
allPosts.push(...filteredPosts);
|
||||
}
|
||||
|
||||
// Sort by createdAt descending and dedupe by id
|
||||
const seen = new Set<string>();
|
||||
const uniquePosts = allPosts
|
||||
.sort((a, b) => new Date(b.createdAt).getTime() - new Date(a.createdAt).getTime())
|
||||
.filter(post => {
|
||||
const key = `${post.nodeDomain}:${post.id}`;
|
||||
if (seen.has(key)) return false;
|
||||
seen.add(key);
|
||||
return true;
|
||||
});
|
||||
|
||||
return {
|
||||
posts: uniquePosts,
|
||||
sources,
|
||||
fetchedAt: new Date().toISOString(),
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,128 @@
|
||||
/**
|
||||
* Swarm Types
|
||||
*
|
||||
* Type definitions for the Synapsis swarm network.
|
||||
*/
|
||||
|
||||
export interface SwarmNodeInfo {
|
||||
domain: string;
|
||||
name?: string;
|
||||
description?: string;
|
||||
logoUrl?: string;
|
||||
publicKey?: string;
|
||||
softwareVersion?: string;
|
||||
userCount?: number;
|
||||
postCount?: number;
|
||||
capabilities?: SwarmCapability[];
|
||||
isNsfw?: boolean;
|
||||
lastSeenAt?: string;
|
||||
}
|
||||
|
||||
export type SwarmCapability = 'handles' | 'gossip' | 'relay' | 'search';
|
||||
|
||||
export interface SwarmAnnouncement {
|
||||
domain: string;
|
||||
name: string;
|
||||
description?: string;
|
||||
logoUrl?: string;
|
||||
publicKey: string;
|
||||
softwareVersion: string;
|
||||
userCount: number;
|
||||
postCount: number;
|
||||
capabilities: SwarmCapability[];
|
||||
isNsfw: boolean;
|
||||
timestamp: string;
|
||||
signature?: string; // Signed with node's private key
|
||||
}
|
||||
|
||||
export interface SwarmGossipPayload {
|
||||
// The node sending this gossip
|
||||
sender: string;
|
||||
|
||||
// Nodes this sender knows about
|
||||
nodes: SwarmNodeInfo[];
|
||||
|
||||
// Optional: handles to sync (piggyback on gossip)
|
||||
handles?: {
|
||||
handle: string;
|
||||
did: string;
|
||||
nodeDomain: string;
|
||||
updatedAt?: string;
|
||||
}[];
|
||||
|
||||
// Timestamp for freshness
|
||||
timestamp: string;
|
||||
|
||||
// Since parameter for incremental sync
|
||||
since?: string;
|
||||
}
|
||||
|
||||
export interface SwarmGossipResponse {
|
||||
// Nodes we're sharing back
|
||||
nodes: SwarmNodeInfo[];
|
||||
|
||||
// Handles we're sharing back
|
||||
handles?: {
|
||||
handle: string;
|
||||
did: string;
|
||||
nodeDomain: string;
|
||||
updatedAt?: string;
|
||||
}[];
|
||||
|
||||
// Stats about what we received
|
||||
received: {
|
||||
nodes: number;
|
||||
handles: number;
|
||||
};
|
||||
}
|
||||
|
||||
export interface SwarmSyncResult {
|
||||
success: boolean;
|
||||
nodesReceived: number;
|
||||
nodesSent: number;
|
||||
handlesReceived: number;
|
||||
handlesSent: number;
|
||||
error?: string;
|
||||
durationMs: number;
|
||||
}
|
||||
|
||||
export interface SwarmStats {
|
||||
totalNodes: number;
|
||||
activeNodes: number;
|
||||
totalUsers: number;
|
||||
totalPosts: number;
|
||||
lastUpdated: string;
|
||||
}
|
||||
|
||||
// Default seed nodes for bootstrapping
|
||||
export const DEFAULT_SEED_NODES = [
|
||||
'node.synapsis.social',
|
||||
] as const;
|
||||
|
||||
// Swarm configuration
|
||||
export const SWARM_CONFIG = {
|
||||
// How often to run gossip (in ms)
|
||||
gossipIntervalMs: 5 * 60 * 1000, // 5 minutes
|
||||
|
||||
// How many nodes to gossip with per round
|
||||
gossipFanout: 3,
|
||||
|
||||
// Max nodes to include in a single gossip message
|
||||
maxNodesPerGossip: 100,
|
||||
|
||||
// Max handles to include in a single gossip message
|
||||
maxHandlesPerGossip: 500,
|
||||
|
||||
// How long before a node is considered inactive
|
||||
inactiveThresholdMs: 24 * 60 * 60 * 1000, // 24 hours
|
||||
|
||||
// How many consecutive failures before marking inactive
|
||||
maxConsecutiveFailures: 5,
|
||||
|
||||
// Trust score adjustments
|
||||
trustScoreOnSuccess: 1,
|
||||
trustScoreOnFailure: -5,
|
||||
minTrustScore: 0,
|
||||
maxTrustScore: 100,
|
||||
defaultTrustScore: 50,
|
||||
} as const;
|
||||
Reference in New Issue
Block a user