/** * Abstracts where inbound interactions come from, selected by `FARCASTER_MODE`. * `FarcasterPollingSource` fetches mentions/replies on `FARCASTER_POLL_INTERVAL` * and dedupes against existing memories before dispatching; `FarcasterWebhookSource` * stays passive and only forwards payloads delivered to the `/webhook` route. Both * feed the shared `IInteractionProcessor`; `createFarcasterInteractionSource` picks * one (polling is the default). */ import type { IAgentRuntime } from "@elizaos/core"; import type { FarcasterClient } from "../client/FarcasterClient"; import type { FarcasterConfig, NeynarWebhookData } from "../types"; import { castUuid, neynarCastToCast } from "../utils"; import type { IInteractionProcessor } from "./InteractionProcessor"; interface FarcasterInteractionSourceParams { client: FarcasterClient; runtime: IAgentRuntime; config: FarcasterConfig; processor: IInteractionProcessor; } export abstract class FarcasterInteractionSource { protected client: FarcasterClient; protected runtime: IAgentRuntime; protected config: FarcasterConfig; protected processor: IInteractionProcessor; protected isRunning: boolean = false; constructor(params: FarcasterInteractionSourceParams) { this.client = params.client; this.runtime = params.runtime; this.config = params.config; this.processor = params.processor; } abstract start(): Promise; abstract stop(): Promise; } /** * Polling-based interaction source. */ export class FarcasterPollingSource extends FarcasterInteractionSource { private timeout: ReturnType | undefined; async start(): Promise { this.runtime.logger.info("Starting Farcaster polling mode"); if (this.isRunning) { return; } this.isRunning = true; void this.runPeriodically(); } async stop(): Promise { this.runtime.logger.info("Stopping Farcaster polling mode"); if (this.timeout) clearTimeout(this.timeout); this.isRunning = false; } private async runPeriodically(): Promise { while (this.isRunning) { try { await this.pollForInteractions(); const delay = this.config.FARCASTER_POLL_INTERVAL * 1000; await new Promise((resolve) => { this.timeout = setTimeout(resolve, delay); }); } catch (error) { this.runtime.logger.error({ error }, "[Farcaster] Error in polling:"); } } } private async pollForInteractions(): Promise { const agentFid = this.config.FARCASTER_FID; const mentions = await this.client.getMentions({ fid: agentFid, pageSize: 20, }); for (const cast of mentions) { try { const mention = neynarCastToCast(cast); const memoryId = castUuid({ agentId: this.runtime.agentId, hash: mention.hash, }); if (await this.runtime.getMemoryById(memoryId)) { continue; } this.runtime.logger.info({ hash: mention.hash }, "New Cast found"); if (mention.authorFid === agentFid) { const memory = await this.processor.ensureCastConnection(mention); await this.runtime.addEmbeddingToMemory(memory); await this.runtime.createMemory(memory, "messages"); continue; } await this.processor.processMention(cast); } catch (error) { this.runtime.logger.error({ error }, "[Farcaster] Error processing mention:"); } } } } export class FarcasterWebhookSource extends FarcasterInteractionSource { async start(): Promise { this.runtime.logger.info("Starting Farcaster webhook mode"); if (this.isRunning) { return; } this.isRunning = true; this.runtime.logger.info("Webhook source is active - waiting for webhook events"); } async stop(): Promise { this.runtime.logger.info("Stopping Farcaster webhook mode"); this.isRunning = false; } async processWebhookData(webhookData: NeynarWebhookData): Promise { if (!this.isRunning) { this.runtime.logger.warn("Webhook source is not running, ignoring webhook data"); return; } try { await this.processor.processWebhookData(webhookData); } catch (error) { this.runtime.logger.error({ error }, "[Farcaster] Error processing webhook data:"); } } } export function createFarcasterInteractionSource( params: FarcasterInteractionSourceParams ): FarcasterInteractionSource { const mode = params.config.FARCASTER_MODE; switch (mode) { case "webhook": return new FarcasterWebhookSource(params); default: return new FarcasterPollingSource(params); } }