Inputs

import { input, context, type AnyAgent } from "@daydreamsai/core";
import { z } from "zod";
import { EventEmitter } from "events"; // Example: Using a simple Node.js EventEmitter
 
// Assume myContext is defined elsewhere
declare const myContext: any;
 
// Example: An input source listening to a simple EventEmitter
const myEventEmitter = new EventEmitter();
 
const eventInput = input({
  // Required: A unique identifier for this input source type.
  // Helps in logging and debugging. e.g., "myApi:webhook", "discord:message"
  type: "custom:event",
 
  // Optional (but recommended): Zod schema for the *data payload* that will be
  // passed into the agent via the `send` function below. This validates
  // the data structure *before* it triggers the agent's run cycle.
  schema: z.object({
    eventId: z.string().describe("Unique ID for the incoming event"),
    payload: z.any().describe("The actual data content of the event"),
    source: z.string().optional().describe("Origin of the event"),
  }),
 
  // Required (usually): Connects to the external data source and calls `send`
  // when new data arrives.
  subscribe: (send, agent: AnyAgent) => {
    // `send` is the crucial function provided by the framework. You call this
    // to push validated data into the agent for processing.
    // `agent` is the agent instance, useful for accessing shared services or config.
 
    const listener = (eventData: {
      id: string;
      data: any;
      origin?: string;
    }) => {
      console.log(`External event received: ${eventData.id}`);
 
      // 1. Determine the target Context and its arguments.
      // This tells the agent *which* context instance should handle this input.
      // This might be static or dynamically determined based on eventData.
      const targetContext = myContext; // Replace with your actual context definition
      const contextArgs = { someId: eventData.id }; // Args matching targetContext.schema
 
      // 2. Prepare the data payload according to this input's `schema`.
      const inputData = {
        eventId: eventData.id,
        payload: eventData.data,
        source: eventData.origin,
      };
 
      try {
        // 3. Validate the payload before sending (optional but good practice)
        eventInput.schema?.parse(inputData); // Use the defined schema
 
        // 4. Call `send` to push the input data into the agent.
        // This triggers the agent's processing cycle for the target context instance.
        send(targetContext, contextArgs, inputData);
      } catch (validationError) {
        agent.logger.error(
          "eventInput:subscribe",
          "Invalid event data received",
          { eventData, validationError }
        );
        // Decide how to handle invalid data (e.g., log, ignore)
      }
    };
 
    // Attach the listener to the external source
    myEventEmitter.on("newEvent", listener);
    agent.logger.info("eventInput", "Subscribed to newEvent");
 
    // IMPORTANT: Return a cleanup function.
    // This is called when the agent stops, ensuring you detach listeners,
    // close connections, clear intervals, etc.
    return () => {
      myEventEmitter.off("newEvent", listener);
      agent.logger.info("eventInput", "Unsubscribed from newEvent");
    };
  },
 
  // Optional: Pre-process data *after* `send` is called but *before* the log entry
  // (InputRef) is created. Useful for adding metadata based on context state.
  handler: async (data, ctx, agent) => {
    // `data` is the payload passed to `send`.
    // `ctx` is the ContextState of the target context instance.
    // Example: Check if this event was already seen in this context instance
    const seen = ctx.memory.processedEventIds?.includes(data.eventId) ?? false;
    return {
      data: data, // Can optionally transform the data here
      params: { seen: String(seen) }, // Add parameters to the InputRef log entry
    };
  },
 
  // Optional: Customize how this input is represented in logs or prompts.
  format: (ref) => {
    // `ref` is the InputRef object (log entry)
    return `Event Received: ID=${ref.data.eventId}, Source=${ref.data.source ?? "N/A"}`;
  },
 
  // Optional: One-time setup logic when the agent starts.
  install: async (agent) => {
    // E.g., agent.container.resolve('myApiClient').connect();
    agent.logger.info("eventInput", "Install hook executed");
  },
});

import { extension, input /* ... */ } from "@daydreamsai/core";
import { z } from "zod";
import { myService } from "./myService";
import { myContext } from "./myContext";
 
export const myApiExtension = extension({
  name: "myApi",
  services: [myService], // Optional services used by the input
  contexts: { myContext }, // Optional contexts targeted by the input
 
  inputs: {
    // Input definitions go here, keyed by their type
    "myApi:webhook": input({
      type: "myApi:webhook", // Redundant but ensures key matches type
      schema: z.object({
        /* ... webhook payload schema ... */
      }),
      subscribe(send, agent) {
        const apiClient = agent.container.resolve<MyApiClient>("myApiClient"); // Use a service
 
        const webhookHandler = (payload: any) => {
          const validatedData =
            myApiExtension.inputs["myApi:webhook"].schema.parse(payload);
          // Determine context/args based on payload
          const contextArgs = { entityId: validatedData.entityId };
          send(myContext, contextArgs, validatedData);
        };
 
        apiClient.registerWebhookListener(webhookHandler);
        return () => apiClient.removeWebhookListener(webhookHandler);
      },
      // ... other input properties ...
    }),
    // ... potentially other inputs for this extension ...
  },
});

// Simplified example
input({
  type: "cli:input",
  subscribe(send, { container }) {
    const rl = container.resolve<readline.Interface>("readline");
    const listen = async () => {
      while (true) {
        // Basic loop, real implementation handles abort
        const line = await rl.question("> ");
        if (line === "exit") break;
        // Send to a default CLI context instance
        send(cliContext, { user: "cli_user" }, line);
      }
    };
    listen(); // Start listening
    return () => {
      /* Abort logic here */
    };
  },
});

// Simplified example
input({
  type: "telegram:message",
  schema: z.object({
    /* ... */
  }),
  subscribe(send, { container }) {
    const tg = container.resolve<Telegraf>("telegraf");
    tg.on("message", (ctx) => {
      if ("text" in ctx.message) {
        const dataPayload = {
          /* ... extract user, text ... */
        };
        // Send data to the specific telegram chat context
        send(telegramChat, { chatId: ctx.chat.id }, dataPayload);
      }
    });
    // Telegraf handles cleanup
    return () => {};
  },
});