# /console.md

---

title: Console
summary: Server-Sent Events (SSE) allow you to stream intermediate results to the client in real-time while the serverless function is processing.
layout: base_layout.njk
date: Last Modified
show_last_updated: true
console: true

---

[How it works](/dev/server-side-events/)

<div class="container-fluid">
    <div class="row">
      <div class="col-3 mb-3 form-floating">
        <select id="function-select" class="form-select">
            <option value="simple">simple</option>
            <option value="xata">xata</option>
            <option value="retrievalAugmentedGeneration">retrievalAugmentedGeneration</option>
        </select>
        <label for="function-select">Serverless Function</label>
        <textarea id="parameters" class="form-control shadow p-3 mb-5 bg-body-tertiary rounded" aria-label="Parameters" aria-describedby="parameters"></textarea>
        <div class="form-check">
            <input class="form-check-input" type="checkbox" value="" id="chkstreamEventsToClient">
            <label class="form-check-label" for="chkstreamEventsToClient">Stream Events</label>
        </div>
        <button id="btnSubmit" type="submit" class="btn btn-primary">Send</button>
      </div>
      <div class="col">
        <div id="streamed-data-results" class="col mx-auto text-success shadow p-3 mb-5 bg-body-tertiary rounded"></div>
      </div>  
      <div class="col">
        <div id="streamed-data-events" class="col mx-auto console shadow p-3 mb-5 rounded d-none"></div>
      </div>
    </div>
</div>

<!-- Hidden JSON element -->
<div id="hidden-json" style="display: none;">
    {
        "simple": {
            "parameters": { "topic": "cats" },
            "expecting": "cats"
        },
        "xata": {
            "parameters": { "orgId": "toip" },
            "expecting": "messages_toip"
        },
        "retrievalAugmentedGeneration": {
            "parameters": { "orgId": "toip", "query": "Who lead the effort on the Issuer Requirements Guide according to the GSWG 2024-01-11 transcripts?" },
            "expecting": "Scott Perry"
        }
    }
</div>

// consoleClient.js
const btnSubmit = document.getElementById("btnSubmit");
const dataDivEvents = document.getElementById("streamed-data-events");
const dataDivResults = document.getElementById("streamed-data-results");
const functionSelect = document.getElementById("function-select");
const parametersTextarea = document.getElementById("parameters");
const hiddenJson = document.getElementById("hidden-json").textContent;
const jsonData = JSON.parse(hiddenJson);
const chkstreamEventsToClient = document.getElementById("chkstreamEventsToClient");

functionSelect.addEventListener("change", function () {
  const testFunc = functionSelect.value;
  const testParms = jsonData[testFunc].parameters;
  parametersTextarea.value = JSON.stringify(testParms, null, 2);
  adjustTextareaHeight();
});

// Trigger change event on page load to populate textarea with initial value
functionSelect.dispatchEvent(new Event("change"));

// Adjust the height of the textarea to fit the content
function adjustTextareaHeight() {
  parametersTextarea.style.height = "auto"; // Reset the height
  parametersTextarea.style.height =
    Math.max(parametersTextarea.scrollHeight, 100) + "px"; // Set the new height with a minimum of 38px (or any other value you prefer)
}

// Adjust the height initially and on input events
adjustTextareaHeight();
parametersTextarea.addEventListener("input", adjustTextareaHeight);

// Checkbox to Stream Events
chkstreamEventsToClient.addEventListener("change", () => {
  if (chkstreamEventsToClient.checked) {
    // Remove class 'd-none' to show the div for events
    dataDivEvents.classList.remove("d-none");
  } else {
    // Toggle visibility of the div for events
    dataDivEvents.classList.add("d-none");
  }
});

// Handle the form submission
async function handlEdgeFunction() {
  const testFunc = functionSelect.value;
  const testParms = JSON.parse(parametersTextarea.value);
  const streamEventsToClient = chkstreamEventsToClient.checked;
  const selectedTest = { testFunc, testParms, streamEventsToClient };
  const url = `${window.location.origin}/simple`;

  try {
    dataDivEvents.innerText = ""; // Clear the div before streaming new data
    dataDivResults.innerText = ""; // Clear the div before streaming new data
    const response = await fetch(url, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
      },
      body: JSON.stringify(selectedTest),
    });
    console.log(`consoleClient url ${url} response.status: ${response.status}`);
    if (!response.ok) {
      throw new Error("Failed to get response from server");
    }

    const reader = response.body.getReader();
    const decoder = new TextDecoder("utf-8");
    let content = "";

    try {
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;

        content += decoder.decode(value, { stream: true });

        // Split the content by the SSE message separator
        const messages = content.split("\n\n");
        const isSSE = content.includes("data: ");

        for (const message of messages.slice(0, -1)) {
          // Process all messages except the last (incomplete) one

          // Server-sent events
          if (isSSE) {
            const data = message.replace("data: ", ""); // Remove the "data: " prefix
            try {
              const parsedData = JSON.parse(data);
              // Update the div for events
              if (parsedData.type === "events-llm") {
                dataDivEvents.innerText += JSON.stringify(
                  parsedData.llm,
                  null,
                  2
                );
              } else if (parsedData.type === "events-prompts") {
                dataDivEvents.innerText += JSON.stringify(
                  parsedData.prompts,
                  null,
                  2
                );
              } else if (parsedData.type === "events-output") {
                dataDivEvents.innerText += JSON.stringify(
                  parsedData.output,
                  null,
                  2
                );
              } else if (parsedData.type === "events-err") {
                dataDivEvents.innerText += JSON.stringify(
                  parsedData.err,
                  null,
                  2
                );

                // Update the div for results
              } else if (parsedData.type === "results") {
                dataDivResults.innerText += parsedData.chunk;
              } else {
                console.error("Unknown message type:", parsedData.type);
              }
            } catch (error) {
              console.error("Error parsing JSON:", error);
            }
          } else {
            // Regular response

            // Update the div for results
            dataDivResults.innerText += message;
          } // isSSE
        } // for message
        content = messages[messages.length - 1]; // Keep the last (incomplete) message for the next iteration
        // } catch (error) {
        //   console.error("Error reading stream:", error);
        //   // Handle stream reading error...
        // } finally {
        //   reader.cancel(); // Close the reader
      }
    } catch (error) {
      console.error("Error reading stream:", error);
      // Handle stream reading error...
    } finally {
      reader.cancel(); // Close the reader
    }
  } catch (error) {
    console.error("Error:", error);
  }
}

btnSubmit.addEventListener("click", async function (event) {
  event.preventDefault();
  handlEdgeFunction();
});

// /edge/simple_Edge_test.js
export default async (request, context) => {
  const funcName = "simple_Edge_test";
  let protocol, body, testFunc, testParms;

  try {
    if (request.method == "GET") {
      // Create a URL object from the request URL
      const url = new URL(request.url);
      // Use the searchParams property to get specific query parameters
      testFunc = url.searchParams.get("testFunc");
      testParms = url.searchParams.get("testParms");
    } else if (request.method == "POST") {
      body = await request.json();
      testFunc = body.testFunc;
      testParms = body.testParms;
    } else {
      return new Response(`${funcName} Invalid Method`, { status: 405 });
    }
    if (!testFunc || !testParms) {
      return new Response(`${funcName} Missing testFunc or testParms`, {
        status: 400,
      });
    }

    // Servderless function URL
    const host = request.headers.get("host");
    if (host.includes("localhost")) {
      protocol = "http";
    } else {
      protocol = "https";
    }

    const serverlessFunctionUrl = `${protocol}:${host}/.netlify/functions/${testFunc}_Serverless_test`;

    console.log(`Testing ${testFunc}  ${JSON.stringify(testParms, null, 2)}`);

    // Call the serverless function
    const response = await fetch(serverlessFunctionUrl, {
      method: "POST",
      body: JSON.stringify(body),
      headers: {
        "Content-Type": "application/json",
      },
    });

    console.log(`${testFunc} response.statusCode: ${response.status}`);

    // Handle the response
    // Simply forwards the serverless function's response
    return new Response(response.body, {
      status: response.statusCode,
      headers: response.headers,
    });
  } catch (error) {
    console.error("Error:", error);
    return new Response(`${funcName} Internal Server Error`, { status: 500 });
  }
};

export const config = {
  path: "/simple",
};

// /serverless/simple_Serverless_test.js
import { stream } from "@netlify/functions";
import { StringOutputParser } from "@langchain/core/output_parsers";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { createChatOpenAIModel } from "../utils/helpers.js";
import {
  createResponseStream,
  writeToStream,
  closeStream,
} from "../utils/eventsModule";
export const handler = stream(async (event) => {
  const funcName = "simple_Serverless_test";
  let testFunc,
    testParms,
    streamEventsToClient = false;
  verbose = true;
  console.log(`Testing ${funcName} `);

  try {
    if (event.httpMethod == "POST") {
      const body = JSON.parse(event.body);
      console.log(`Testing ${funcName} body: ${JSON.stringify(body, null, 2)}`);
      testFunc = body.testFunc;
      testParms = body.testParms;
      streamEventsToClient = body.streamEventsToClient;
    } else if (event.httpMethod === "GET") {
      testFunc = event.queryStringParameters.testFunc;
      testParms = event.queryStringParameters.testParms;
      streamEventsToClient = event.queryStringParameters.streamEventsToClient;
    } else {
      return new Response(`${funcName} Invalid Method`, { statusCode: 405 });
    }
    if (!testFunc || !testParms) {
      return new Response(`${funcName} Missing testFunc or testParms`, {
        statusCode: 400,
      });
    }

    // Echo the testFunc, testParms & streamEventsToClient
    console.log(
      `${funcName} testFunc: ${testFunc} streamEventsToClient: ${streamEventsToClient} testParms: ${JSON.stringify(
        testParms,
        null,
        2
      )}`
    );

    // Unique parms to test
    const topic = testParms?.topic;
    if (!topic) {
      return new Response(`${funcName} Missing testParms: topic`, {
        statusCode: 400,
      });
    }
    console.log(`${funcName} testFunc: ${testFunc} topic: ${topic}`);

    // ReadaableStream instance
    //const responseStream = createResponseStream();
    let chunks = "";
    const responseBody = new ReadableStream({
      async start(controller) {
        // Create the OpenAI model
        const options = {
          model: process.env.OPENAI_DEFAULT_MODEL,
          subscribeToEvents: true,
          streamEventsToClient,
          onEvent: (event) => {
            controller.enqueue(`data: ${JSON.stringify(event)}\n\n`);
          },
        };
        const model = await createChatOpenAIModel(options);
        const prompt = ChatPromptTemplate.fromTemplate(
          "Write a short poem about {topic}"
        );
        const parser = new StringOutputParser();
        const chain = prompt.pipe(model).pipe(parser);
        const response = await chain.stream({ topic });

        for await (const chunk of response) {
          controller.enqueue(
            `data: ${JSON.stringify({ type: "results", chunk })}\n\n`
          );
          chunks += chunk;
        }
        controller.close();
      },
    });

    // SSE
    // let chunks = "",
    //   responseBody;
    // const useSSE = false; // is the only way it streams data in real time
    // if (useSSE) {
    //   for await (const chunk of response) {
    //     writeToStream(responseStream, { type: "results", chunk }, verbose);
    //     chunks += chunk;
    //   }

    //   // Remove all listeners from the stream before returning it
    //   closeStream(responseStream, verbose);
    // } else {
    //   responseBody = new ReadableStream({
    //     async start(controller) {
    //       for await (const chunk of response) {
    //         controller.enqueue(
    //           `data: ${JSON.stringify({ type: "results", chunk })}\n\n`
    //         );
    //         chunks += chunk;
    //       }
    //       controller.close();
    //     },
    //   });
    // }

    console.log(`simple_Serverless_test chunks: ${chunks}`);

    return {
      headers: {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache",
//        Connection: "keep-alive",
      },
      statusCode: 200,
      body: responseBody, // useSSE ? responseStream : responseBody,
    };
  } catch (error) {
    console.error("error", error);
    return {
      statusCode: 500,
      body: JSON.stringify({
        error: `${funcName} Error calling ${testFunc} with parms ${testParms}`,
      }),
    };
  }
});

// helpers.js function

/**
 * @name createChatOpenAIModel
 * @description Create Chat OpenAI Model
 * @param {object} options
 * @param {string} options.modelName
 * @param {string} options.openAIApiKey
 * @param {string} options.temperature
 * @param {string} options.streaming
 * @param {object} options.signal
 * @param {number} [options.timeout] - 1000 * 60 * 1;   // 1 minutes ms * seconds * minutes
 * @param {number} [options.maxRetries] - 10;           // Dealing with API Errors
 * @param {number} [options.maxConcurrency] - 5;        // Deal with API Rate Limits
 * @param {boolean} [options.subscribeToEvents] - false // Subscribe to events
 * @param {object} [options.streamEventsToClient]               //  if true stream for events
 * @param {object} [options.onEvent]                    //  function to eenqueue events
 * @param {boolean} [options.intermediateSteps] - false // Intermediate steps
 * @param {boolean} [options.verbose] - env "VERBOSE"
 * @return {Promise