Fix hero card aspect ratio and add horoscope generation functionality
Introduce horoscope generation via OpenAI API, including new API endpoints and database schema. Adjust card components in `home.tsx` to use `aspect-[16/9]` for consistent image sizing, resolving previous height stretching issues. Update dependencies in `package.json`. Replit-Commit-Author: Agent Replit-Commit-Session-Id: 413891e8-d784-4bea-b9f5-91a5a68316b4 Replit-Commit-Checkpoint-Type: full_checkpoint Replit-Commit-Event-Id: ca1aa952-242c-43c1-9e28-47aed39cee1b Replit-Commit-Screenshot-Url: https://storage.googleapis.com/screenshot-production-us-central1/f209e72a-0939-48fa-84fc-57854de71967/413891e8-d784-4bea-b9f5-91a5a68316b4/nTLKCC5 Replit-Helium-Checkpoint-Created: true
This commit is contained in:
@@ -0,0 +1,274 @@
|
||||
import OpenAI, { toFile } from "openai";
|
||||
import { Buffer } from "node:buffer";
|
||||
import { spawn } from "child_process";
|
||||
import { writeFile, unlink, readFile } from "fs/promises";
|
||||
import { randomUUID } from "crypto";
|
||||
import { tmpdir } from "os";
|
||||
import { join } from "path";
|
||||
|
||||
export const openai = new OpenAI({
|
||||
apiKey: process.env.AI_INTEGRATIONS_OPENAI_API_KEY,
|
||||
baseURL: process.env.AI_INTEGRATIONS_OPENAI_BASE_URL,
|
||||
});
|
||||
|
||||
export type AudioFormat = "wav" | "mp3" | "webm" | "mp4" | "ogg" | "unknown";
|
||||
|
||||
/**
|
||||
* Detect audio format from buffer magic bytes.
|
||||
* Supports: WAV, MP3, WebM (Chrome/Firefox), MP4/M4A/MOV (Safari/iOS), OGG
|
||||
*/
|
||||
export function detectAudioFormat(buffer: Buffer): AudioFormat {
|
||||
if (buffer.length < 12) return "unknown";
|
||||
|
||||
// WAV: RIFF....WAVE
|
||||
if (buffer[0] === 0x52 && buffer[1] === 0x49 && buffer[2] === 0x46 && buffer[3] === 0x46) {
|
||||
return "wav";
|
||||
}
|
||||
// WebM: EBML header
|
||||
if (buffer[0] === 0x1a && buffer[1] === 0x45 && buffer[2] === 0xdf && buffer[3] === 0xa3) {
|
||||
return "webm";
|
||||
}
|
||||
// MP3: ID3 tag or frame sync
|
||||
if (
|
||||
(buffer[0] === 0xff && (buffer[1] === 0xfb || buffer[1] === 0xfa || buffer[1] === 0xf3)) ||
|
||||
(buffer[0] === 0x49 && buffer[1] === 0x44 && buffer[2] === 0x33)
|
||||
) {
|
||||
return "mp3";
|
||||
}
|
||||
// MP4/M4A/MOV: ....ftyp (Safari/iOS records in these containers)
|
||||
if (buffer[4] === 0x66 && buffer[5] === 0x74 && buffer[6] === 0x79 && buffer[7] === 0x70) {
|
||||
return "mp4";
|
||||
}
|
||||
// OGG: OggS
|
||||
if (buffer[0] === 0x4f && buffer[1] === 0x67 && buffer[2] === 0x67 && buffer[3] === 0x53) {
|
||||
return "ogg";
|
||||
}
|
||||
return "unknown";
|
||||
}
|
||||
|
||||
/**
|
||||
* Convert any audio/video format to WAV using ffmpeg.
|
||||
* Uses temp files instead of pipes because video containers (MP4/MOV)
|
||||
* require seeking to find the audio track.
|
||||
*/
|
||||
export async function convertToWav(audioBuffer: Buffer): Promise<Buffer> {
|
||||
const inputPath = join(tmpdir(), `input-${randomUUID()}`);
|
||||
const outputPath = join(tmpdir(), `output-${randomUUID()}.wav`);
|
||||
|
||||
try {
|
||||
// Write input to temp file (required for video containers that need seeking)
|
||||
await writeFile(inputPath, audioBuffer);
|
||||
|
||||
// Run ffmpeg with file paths
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
const ffmpeg = spawn("ffmpeg", [
|
||||
"-i", inputPath,
|
||||
"-vn", // Extract audio only (ignore video track)
|
||||
"-f", "wav",
|
||||
"-ar", "16000", // 16kHz sample rate (good for speech)
|
||||
"-ac", "1", // Mono
|
||||
"-acodec", "pcm_s16le",
|
||||
"-y", // Overwrite output
|
||||
outputPath,
|
||||
]);
|
||||
|
||||
ffmpeg.stderr.on("data", () => {}); // Suppress logs
|
||||
ffmpeg.on("close", (code) => {
|
||||
if (code === 0) resolve();
|
||||
else reject(new Error(`ffmpeg exited with code ${code}`));
|
||||
});
|
||||
ffmpeg.on("error", reject);
|
||||
});
|
||||
|
||||
// Read converted audio
|
||||
return await readFile(outputPath);
|
||||
} finally {
|
||||
// Clean up temp files
|
||||
await unlink(inputPath).catch(() => {});
|
||||
await unlink(outputPath).catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Auto-detect and convert audio to OpenAI-compatible format.
|
||||
* - WAV/MP3: Pass through (already compatible)
|
||||
* - WebM/MP4/OGG: Convert to WAV via ffmpeg
|
||||
*/
|
||||
export async function ensureCompatibleFormat(
|
||||
audioBuffer: Buffer
|
||||
): Promise<{ buffer: Buffer; format: "wav" | "mp3" }> {
|
||||
const detected = detectAudioFormat(audioBuffer);
|
||||
if (detected === "wav") return { buffer: audioBuffer, format: "wav" };
|
||||
if (detected === "mp3") return { buffer: audioBuffer, format: "mp3" };
|
||||
// Convert WebM, MP4, OGG, or unknown to WAV
|
||||
const wavBuffer = await convertToWav(audioBuffer);
|
||||
return { buffer: wavBuffer, format: "wav" };
|
||||
}
|
||||
|
||||
/**
|
||||
* Voice Chat: User speaks, LLM responds with audio (audio-in, audio-out).
|
||||
* Uses gpt-audio model via Replit AI Integrations.
|
||||
* Note: Browser records WebM/opus - convert to WAV using ffmpeg before calling this.
|
||||
*/
|
||||
export async function voiceChat(
|
||||
audioBuffer: Buffer,
|
||||
voice: "alloy" | "echo" | "fable" | "onyx" | "nova" | "shimmer" = "alloy",
|
||||
inputFormat: "wav" | "mp3" = "wav",
|
||||
outputFormat: "wav" | "mp3" = "mp3"
|
||||
): Promise<{ transcript: string; audioResponse: Buffer }> {
|
||||
const audioBase64 = audioBuffer.toString("base64");
|
||||
const response = await openai.chat.completions.create({
|
||||
model: "gpt-audio",
|
||||
modalities: ["text", "audio"],
|
||||
audio: { voice, format: outputFormat },
|
||||
messages: [{
|
||||
role: "user",
|
||||
content: [
|
||||
{ type: "input_audio", input_audio: { data: audioBase64, format: inputFormat } },
|
||||
],
|
||||
}],
|
||||
});
|
||||
const message = response.choices[0]?.message as any;
|
||||
const transcript = message?.audio?.transcript || message?.content || "";
|
||||
const audioData = message?.audio?.data ?? "";
|
||||
return {
|
||||
transcript,
|
||||
audioResponse: Buffer.from(audioData, "base64"),
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Streaming Voice Chat: For real-time audio responses.
|
||||
* Note: Streaming only supports pcm16 output format.
|
||||
*
|
||||
* @example
|
||||
* // Converting browser WebM to WAV before calling:
|
||||
* const webmBuffer = Buffer.from(req.body.audio, "base64");
|
||||
* const wavBuffer = await convertWebmToWav(webmBuffer);
|
||||
* for await (const chunk of voiceChatStream(wavBuffer)) { ... }
|
||||
*/
|
||||
export async function voiceChatStream(
|
||||
audioBuffer: Buffer,
|
||||
voice: "alloy" | "echo" | "fable" | "onyx" | "nova" | "shimmer" = "alloy",
|
||||
inputFormat: "wav" | "mp3" = "wav"
|
||||
): Promise<AsyncIterable<{ type: "transcript" | "audio"; data: string }>> {
|
||||
const audioBase64 = audioBuffer.toString("base64");
|
||||
const stream = await openai.chat.completions.create({
|
||||
model: "gpt-audio",
|
||||
modalities: ["text", "audio"],
|
||||
audio: { voice, format: "pcm16" },
|
||||
messages: [{
|
||||
role: "user",
|
||||
content: [
|
||||
{ type: "input_audio", input_audio: { data: audioBase64, format: inputFormat } },
|
||||
],
|
||||
}],
|
||||
stream: true,
|
||||
});
|
||||
|
||||
return (async function* () {
|
||||
for await (const chunk of stream) {
|
||||
const delta = chunk.choices?.[0]?.delta as any;
|
||||
if (!delta) continue;
|
||||
if (delta?.audio?.transcript) {
|
||||
yield { type: "transcript", data: delta.audio.transcript };
|
||||
}
|
||||
if (delta?.audio?.data) {
|
||||
yield { type: "audio", data: delta.audio.data };
|
||||
}
|
||||
}
|
||||
})();
|
||||
}
|
||||
|
||||
/**
|
||||
* Text-to-Speech: Converts text to speech verbatim.
|
||||
* Uses gpt-audio model via Replit AI Integrations.
|
||||
*/
|
||||
export async function textToSpeech(
|
||||
text: string,
|
||||
voice: "alloy" | "echo" | "fable" | "onyx" | "nova" | "shimmer" = "alloy",
|
||||
format: "wav" | "mp3" | "flac" | "opus" | "pcm16" = "wav"
|
||||
): Promise<Buffer> {
|
||||
const response = await openai.chat.completions.create({
|
||||
model: "gpt-audio",
|
||||
modalities: ["text", "audio"],
|
||||
audio: { voice, format },
|
||||
messages: [
|
||||
{ role: "system", content: "You are an assistant that performs text-to-speech." },
|
||||
{ role: "user", content: `Repeat the following text verbatim: ${text}` },
|
||||
],
|
||||
});
|
||||
const audioData = (response.choices[0]?.message as any)?.audio?.data ?? "";
|
||||
return Buffer.from(audioData, "base64");
|
||||
}
|
||||
|
||||
/**
|
||||
* Streaming Text-to-Speech: Converts text to speech with real-time streaming.
|
||||
* Uses gpt-audio model via Replit AI Integrations.
|
||||
* Note: Streaming only supports pcm16 output format.
|
||||
*/
|
||||
export async function textToSpeechStream(
|
||||
text: string,
|
||||
voice: "alloy" | "echo" | "fable" | "onyx" | "nova" | "shimmer" = "alloy"
|
||||
): Promise<AsyncIterable<string>> {
|
||||
const stream = await openai.chat.completions.create({
|
||||
model: "gpt-audio",
|
||||
modalities: ["text", "audio"],
|
||||
audio: { voice, format: "pcm16" },
|
||||
messages: [
|
||||
{ role: "system", content: "You are an assistant that performs text-to-speech." },
|
||||
{ role: "user", content: `Repeat the following text verbatim: ${text}` },
|
||||
],
|
||||
stream: true,
|
||||
});
|
||||
|
||||
return (async function* () {
|
||||
for await (const chunk of stream) {
|
||||
const delta = chunk.choices?.[0]?.delta as any;
|
||||
if (!delta) continue;
|
||||
if (delta?.audio?.data) {
|
||||
yield delta.audio.data;
|
||||
}
|
||||
}
|
||||
})();
|
||||
}
|
||||
|
||||
/**
|
||||
* Speech-to-Text: Transcribes audio using dedicated transcription model.
|
||||
* Uses gpt-4o-mini-transcribe for accurate transcription.
|
||||
*/
|
||||
export async function speechToText(
|
||||
audioBuffer: Buffer,
|
||||
format: "wav" | "mp3" | "webm" = "wav"
|
||||
): Promise<string> {
|
||||
const file = await toFile(audioBuffer, `audio.${format}`);
|
||||
const response = await openai.audio.transcriptions.create({
|
||||
file,
|
||||
model: "gpt-4o-mini-transcribe",
|
||||
});
|
||||
return response.text;
|
||||
}
|
||||
|
||||
/**
|
||||
* Streaming Speech-to-Text: Transcribes audio with real-time streaming.
|
||||
* Uses gpt-4o-mini-transcribe for accurate transcription.
|
||||
*/
|
||||
export async function speechToTextStream(
|
||||
audioBuffer: Buffer,
|
||||
format: "wav" | "mp3" | "webm" = "wav"
|
||||
): Promise<AsyncIterable<string>> {
|
||||
const file = await toFile(audioBuffer, `audio.${format}`);
|
||||
const stream = await openai.audio.transcriptions.create({
|
||||
file,
|
||||
model: "gpt-4o-mini-transcribe",
|
||||
stream: true,
|
||||
});
|
||||
|
||||
return (async function* () {
|
||||
for await (const event of stream) {
|
||||
if (event.type === "transcript.text.delta") {
|
||||
yield event.delta;
|
||||
}
|
||||
}
|
||||
})();
|
||||
}
|
||||
@@ -0,0 +1,14 @@
|
||||
export { registerAudioRoutes } from "./routes";
|
||||
export {
|
||||
openai,
|
||||
detectAudioFormat,
|
||||
convertToWav,
|
||||
ensureCompatibleFormat,
|
||||
type AudioFormat,
|
||||
voiceChat,
|
||||
voiceChatStream,
|
||||
textToSpeech,
|
||||
textToSpeechStream,
|
||||
speechToText,
|
||||
speechToTextStream,
|
||||
} from "./client";
|
||||
@@ -0,0 +1,136 @@
|
||||
import express, { type Express, type Request, type Response } from "express";
|
||||
import { chatStorage } from "../chat/storage";
|
||||
import { openai, speechToText, ensureCompatibleFormat } from "./client";
|
||||
|
||||
// Body parser with 50MB limit for audio payloads
|
||||
const audioBodyParser = express.json({ limit: "50mb" });
|
||||
|
||||
export function registerAudioRoutes(app: Express): void {
|
||||
// Get all conversations
|
||||
app.get("/api/conversations", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const conversations = await chatStorage.getAllConversations();
|
||||
res.json(conversations);
|
||||
} catch (error) {
|
||||
console.error("Error fetching conversations:", error);
|
||||
res.status(500).json({ error: "Failed to fetch conversations" });
|
||||
}
|
||||
});
|
||||
|
||||
// Get single conversation with messages
|
||||
app.get("/api/conversations/:id", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const id = parseInt(req.params.id);
|
||||
const conversation = await chatStorage.getConversation(id);
|
||||
if (!conversation) {
|
||||
return res.status(404).json({ error: "Conversation not found" });
|
||||
}
|
||||
const messages = await chatStorage.getMessagesByConversation(id);
|
||||
res.json({ ...conversation, messages });
|
||||
} catch (error) {
|
||||
console.error("Error fetching conversation:", error);
|
||||
res.status(500).json({ error: "Failed to fetch conversation" });
|
||||
}
|
||||
});
|
||||
|
||||
// Create new conversation
|
||||
app.post("/api/conversations", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const { title } = req.body;
|
||||
const conversation = await chatStorage.createConversation(title || "New Chat");
|
||||
res.status(201).json(conversation);
|
||||
} catch (error) {
|
||||
console.error("Error creating conversation:", error);
|
||||
res.status(500).json({ error: "Failed to create conversation" });
|
||||
}
|
||||
});
|
||||
|
||||
// Delete conversation
|
||||
app.delete("/api/conversations/:id", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const id = parseInt(req.params.id);
|
||||
await chatStorage.deleteConversation(id);
|
||||
res.status(204).send();
|
||||
} catch (error) {
|
||||
console.error("Error deleting conversation:", error);
|
||||
res.status(500).json({ error: "Failed to delete conversation" });
|
||||
}
|
||||
});
|
||||
|
||||
// Send voice message and get streaming audio response
|
||||
// Auto-detects audio format and converts WebM/MP4/OGG to WAV
|
||||
// Uses gpt-4o-mini-transcribe for STT, gpt-audio for voice response
|
||||
app.post("/api/conversations/:id/messages", audioBodyParser, async (req: Request, res: Response) => {
|
||||
try {
|
||||
const conversationId = parseInt(req.params.id);
|
||||
const { audio, voice = "alloy" } = req.body;
|
||||
|
||||
if (!audio) {
|
||||
return res.status(400).json({ error: "Audio data (base64) is required" });
|
||||
}
|
||||
|
||||
// 1. Auto-detect format and convert to OpenAI-compatible format
|
||||
const rawBuffer = Buffer.from(audio, "base64");
|
||||
const { buffer: audioBuffer, format: inputFormat } = await ensureCompatibleFormat(rawBuffer);
|
||||
|
||||
// 2. Transcribe user audio
|
||||
const userTranscript = await speechToText(audioBuffer, inputFormat);
|
||||
|
||||
// 3. Save user message
|
||||
await chatStorage.createMessage(conversationId, "user", userTranscript);
|
||||
|
||||
// 4. Get conversation history
|
||||
const existingMessages = await chatStorage.getMessagesByConversation(conversationId);
|
||||
const chatHistory = existingMessages.map((m) => ({
|
||||
role: m.role as "user" | "assistant",
|
||||
content: m.content,
|
||||
}));
|
||||
|
||||
// 5. Set up SSE
|
||||
res.setHeader("Content-Type", "text/event-stream");
|
||||
res.setHeader("Cache-Control", "no-cache");
|
||||
res.setHeader("Connection", "keep-alive");
|
||||
|
||||
res.write(`data: ${JSON.stringify({ type: "user_transcript", data: userTranscript })}\n\n`);
|
||||
|
||||
// 6. Stream audio response from gpt-audio
|
||||
const stream = await openai.chat.completions.create({
|
||||
model: "gpt-audio",
|
||||
modalities: ["text", "audio"],
|
||||
audio: { voice, format: "pcm16" },
|
||||
messages: chatHistory,
|
||||
stream: true,
|
||||
});
|
||||
|
||||
let assistantTranscript = "";
|
||||
|
||||
for await (const chunk of stream) {
|
||||
const delta = chunk.choices?.[0]?.delta as any;
|
||||
if (!delta) continue;
|
||||
|
||||
if (delta?.audio?.transcript) {
|
||||
assistantTranscript += delta.audio.transcript;
|
||||
res.write(`data: ${JSON.stringify({ type: "transcript", data: delta.audio.transcript })}\n\n`);
|
||||
}
|
||||
|
||||
if (delta?.audio?.data) {
|
||||
res.write(`data: ${JSON.stringify({ type: "audio", data: delta.audio.data })}\n\n`);
|
||||
}
|
||||
}
|
||||
|
||||
// 7. Save assistant message
|
||||
await chatStorage.createMessage(conversationId, "assistant", assistantTranscript);
|
||||
|
||||
res.write(`data: ${JSON.stringify({ type: "done", transcript: assistantTranscript })}\n\n`);
|
||||
res.end();
|
||||
} catch (error) {
|
||||
console.error("Error processing voice message:", error);
|
||||
if (res.headersSent) {
|
||||
res.write(`data: ${JSON.stringify({ type: "error", error: "Failed to process voice message" })}\n\n`);
|
||||
res.end();
|
||||
} else {
|
||||
res.status(500).json({ error: "Failed to process voice message" });
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
export {
|
||||
batchProcess,
|
||||
batchProcessWithSSE,
|
||||
isRateLimitError,
|
||||
type BatchOptions,
|
||||
} from "./utils";
|
||||
|
||||
@@ -0,0 +1,182 @@
|
||||
import pLimit from "p-limit";
|
||||
import pRetry from "p-retry";
|
||||
|
||||
/**
|
||||
* Batch Processing Utilities
|
||||
*
|
||||
* This module provides a generic batch processing function with built-in
|
||||
* rate limiting and automatic retries. Use it for any task that requires
|
||||
* processing multiple items through an LLM or external API.
|
||||
*
|
||||
* USAGE:
|
||||
* ```typescript
|
||||
* import { batchProcess, isRateLimitError } from "./replit_integrations/batch";
|
||||
*
|
||||
* const results = await batchProcess(
|
||||
* artworks,
|
||||
* async (artwork) => {
|
||||
* // Your custom LLM logic here
|
||||
* const response = await openai.chat.completions.create({
|
||||
* model: "gpt-5.1",
|
||||
* messages: [{ role: "user", content: `Categorize: ${artwork.name}` }],
|
||||
* response_format: { type: "json_object" },
|
||||
* });
|
||||
* return JSON.parse(response.choices[0]?.message?.content || "{}");
|
||||
* },
|
||||
* { concurrency: 2, retries: 5 }
|
||||
* );
|
||||
* ```
|
||||
*/
|
||||
|
||||
export interface BatchOptions {
|
||||
/** Max concurrent requests (default: 2) */
|
||||
concurrency?: number;
|
||||
/** Max retry attempts for rate limit errors (default: 7) */
|
||||
retries?: number;
|
||||
/** Initial retry delay in ms (default: 2000) */
|
||||
minTimeout?: number;
|
||||
/** Max retry delay in ms (default: 128000) */
|
||||
maxTimeout?: number;
|
||||
/** Callback for progress updates */
|
||||
onProgress?: (completed: number, total: number, item: unknown) => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if an error is a rate limit or quota violation.
|
||||
* Use this in custom error handling if needed.
|
||||
*/
|
||||
export function isRateLimitError(error: unknown): boolean {
|
||||
const errorMsg = error instanceof Error ? error.message : String(error);
|
||||
return (
|
||||
errorMsg.includes("429") ||
|
||||
errorMsg.includes("RATELIMIT_EXCEEDED") ||
|
||||
errorMsg.toLowerCase().includes("quota") ||
|
||||
errorMsg.toLowerCase().includes("rate limit")
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* Process items in batches with rate limiting and automatic retries.
|
||||
*
|
||||
* @param items - Array of items to process
|
||||
* @param processor - Async function to process each item (write your LLM logic here)
|
||||
* @param options - Concurrency and retry settings
|
||||
* @returns Promise resolving to array of results in the same order as input
|
||||
*
|
||||
* @example
|
||||
* // Process CSV artwork data with custom categorization
|
||||
* const categorized = await batchProcess(
|
||||
* csvRows,
|
||||
* async (row) => {
|
||||
* const response = await openai.chat.completions.create({
|
||||
* model: "gpt-5.1", // the newest OpenAI model
|
||||
* messages: [{ role: "user", content: `Categorize artwork: ${row.name}` }],
|
||||
* response_format: { type: "json_object" },
|
||||
* });
|
||||
* return { ...row, category: JSON.parse(response.choices[0]?.message?.content || "{}") };
|
||||
* }
|
||||
* );
|
||||
*/
|
||||
export async function batchProcess<T, R>(
|
||||
items: T[],
|
||||
processor: (item: T, index: number) => Promise<R>,
|
||||
options: BatchOptions = {}
|
||||
): Promise<R[]> {
|
||||
const {
|
||||
concurrency = 2,
|
||||
retries = 7,
|
||||
minTimeout = 2000,
|
||||
maxTimeout = 128000,
|
||||
onProgress,
|
||||
} = options;
|
||||
|
||||
const limit = pLimit(concurrency);
|
||||
let completed = 0;
|
||||
|
||||
const promises = items.map((item, index) =>
|
||||
limit(() =>
|
||||
pRetry(
|
||||
async () => {
|
||||
try {
|
||||
const result = await processor(item, index);
|
||||
completed++;
|
||||
onProgress?.(completed, items.length, item);
|
||||
return result;
|
||||
} catch (error: unknown) {
|
||||
if (isRateLimitError(error)) {
|
||||
throw error; // Rethrow to trigger p-retry
|
||||
}
|
||||
// For non-rate-limit errors, abort immediately
|
||||
throw new pRetry.AbortError(
|
||||
error instanceof Error ? error : new Error(String(error))
|
||||
);
|
||||
}
|
||||
},
|
||||
{ retries, minTimeout, maxTimeout, factor: 2 }
|
||||
)
|
||||
)
|
||||
);
|
||||
|
||||
return Promise.all(promises);
|
||||
}
|
||||
|
||||
/**
|
||||
* Process items sequentially with SSE progress streaming.
|
||||
* Use this when you need real-time progress updates to the client.
|
||||
*
|
||||
* @param items - Array of items to process
|
||||
* @param processor - Async function to process each item
|
||||
* @param sendEvent - Function to send SSE events to the client
|
||||
* @param options - Retry settings (concurrency is always 1 for sequential)
|
||||
*/
|
||||
export async function batchProcessWithSSE<T, R>(
|
||||
items: T[],
|
||||
processor: (item: T, index: number) => Promise<R>,
|
||||
sendEvent: (event: { type: string; [key: string]: unknown }) => void,
|
||||
options: Omit<BatchOptions, "concurrency" | "onProgress"> = {}
|
||||
): Promise<R[]> {
|
||||
const { retries = 5, minTimeout = 1000, maxTimeout = 15000 } = options;
|
||||
|
||||
sendEvent({ type: "started", total: items.length });
|
||||
|
||||
const results: R[] = [];
|
||||
let errors = 0;
|
||||
|
||||
for (let index = 0; index < items.length; index++) {
|
||||
const item = items[index];
|
||||
sendEvent({ type: "processing", index, item });
|
||||
|
||||
try {
|
||||
const result = await pRetry(
|
||||
() => processor(item, index),
|
||||
{
|
||||
retries,
|
||||
minTimeout,
|
||||
maxTimeout,
|
||||
factor: 2,
|
||||
onFailedAttempt: (error) => {
|
||||
if (!isRateLimitError(error)) {
|
||||
throw new pRetry.AbortError(
|
||||
error instanceof Error ? error : new Error(String(error))
|
||||
);
|
||||
}
|
||||
},
|
||||
}
|
||||
);
|
||||
results.push(result);
|
||||
sendEvent({ type: "progress", index, result });
|
||||
} catch (error) {
|
||||
errors++;
|
||||
results.push(undefined as R); // Placeholder for failed items
|
||||
sendEvent({
|
||||
type: "progress",
|
||||
index,
|
||||
error: error instanceof Error ? error.message : "Processing failed",
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
sendEvent({ type: "complete", processed: items.length, errors });
|
||||
return results;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
export { registerChatRoutes } from "./routes";
|
||||
export { chatStorage, type IChatStorage } from "./storage";
|
||||
|
||||
@@ -0,0 +1,118 @@
|
||||
import type { Express, Request, Response } from "express";
|
||||
import OpenAI from "openai";
|
||||
import { chatStorage } from "./storage";
|
||||
|
||||
const openai = new OpenAI({
|
||||
apiKey: process.env.AI_INTEGRATIONS_OPENAI_API_KEY,
|
||||
baseURL: process.env.AI_INTEGRATIONS_OPENAI_BASE_URL,
|
||||
});
|
||||
|
||||
export function registerChatRoutes(app: Express): void {
|
||||
// Get all conversations
|
||||
app.get("/api/conversations", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const conversations = await chatStorage.getAllConversations();
|
||||
res.json(conversations);
|
||||
} catch (error) {
|
||||
console.error("Error fetching conversations:", error);
|
||||
res.status(500).json({ error: "Failed to fetch conversations" });
|
||||
}
|
||||
});
|
||||
|
||||
// Get single conversation with messages
|
||||
app.get("/api/conversations/:id", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const id = parseInt(req.params.id);
|
||||
const conversation = await chatStorage.getConversation(id);
|
||||
if (!conversation) {
|
||||
return res.status(404).json({ error: "Conversation not found" });
|
||||
}
|
||||
const messages = await chatStorage.getMessagesByConversation(id);
|
||||
res.json({ ...conversation, messages });
|
||||
} catch (error) {
|
||||
console.error("Error fetching conversation:", error);
|
||||
res.status(500).json({ error: "Failed to fetch conversation" });
|
||||
}
|
||||
});
|
||||
|
||||
// Create new conversation
|
||||
app.post("/api/conversations", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const { title } = req.body;
|
||||
const conversation = await chatStorage.createConversation(title || "New Chat");
|
||||
res.status(201).json(conversation);
|
||||
} catch (error) {
|
||||
console.error("Error creating conversation:", error);
|
||||
res.status(500).json({ error: "Failed to create conversation" });
|
||||
}
|
||||
});
|
||||
|
||||
// Delete conversation
|
||||
app.delete("/api/conversations/:id", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const id = parseInt(req.params.id);
|
||||
await chatStorage.deleteConversation(id);
|
||||
res.status(204).send();
|
||||
} catch (error) {
|
||||
console.error("Error deleting conversation:", error);
|
||||
res.status(500).json({ error: "Failed to delete conversation" });
|
||||
}
|
||||
});
|
||||
|
||||
// Send message and get AI response (streaming)
|
||||
app.post("/api/conversations/:id/messages", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const conversationId = parseInt(req.params.id);
|
||||
const { content } = req.body;
|
||||
|
||||
// Save user message
|
||||
await chatStorage.createMessage(conversationId, "user", content);
|
||||
|
||||
// Get conversation history for context
|
||||
const messages = await chatStorage.getMessagesByConversation(conversationId);
|
||||
const chatMessages = messages.map((m) => ({
|
||||
role: m.role as "user" | "assistant",
|
||||
content: m.content,
|
||||
}));
|
||||
|
||||
// Set up SSE
|
||||
res.setHeader("Content-Type", "text/event-stream");
|
||||
res.setHeader("Cache-Control", "no-cache");
|
||||
res.setHeader("Connection", "keep-alive");
|
||||
|
||||
// Stream response from OpenAI
|
||||
const stream = await openai.chat.completions.create({
|
||||
model: "gpt-5.1",
|
||||
messages: chatMessages,
|
||||
stream: true,
|
||||
max_completion_tokens: 8192,
|
||||
});
|
||||
|
||||
let fullResponse = "";
|
||||
|
||||
for await (const chunk of stream) {
|
||||
const content = chunk.choices[0]?.delta?.content || "";
|
||||
if (content) {
|
||||
fullResponse += content;
|
||||
res.write(`data: ${JSON.stringify({ content })}\n\n`);
|
||||
}
|
||||
}
|
||||
|
||||
// Save assistant message
|
||||
await chatStorage.createMessage(conversationId, "assistant", fullResponse);
|
||||
|
||||
res.write(`data: ${JSON.stringify({ done: true })}\n\n`);
|
||||
res.end();
|
||||
} catch (error) {
|
||||
console.error("Error sending message:", error);
|
||||
// Check if headers already sent (SSE streaming started)
|
||||
if (res.headersSent) {
|
||||
res.write(`data: ${JSON.stringify({ error: "Failed to send message" })}\n\n`);
|
||||
res.end();
|
||||
} else {
|
||||
res.status(500).json({ error: "Failed to send message" });
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
import { db } from "../../db";
|
||||
import { conversations, messages } from "@shared/schema";
|
||||
import { eq, desc } from "drizzle-orm";
|
||||
|
||||
export interface IChatStorage {
|
||||
getConversation(id: number): Promise<typeof conversations.$inferSelect | undefined>;
|
||||
getAllConversations(): Promise<(typeof conversations.$inferSelect)[]>;
|
||||
createConversation(title: string): Promise<typeof conversations.$inferSelect>;
|
||||
deleteConversation(id: number): Promise<void>;
|
||||
getMessagesByConversation(conversationId: number): Promise<(typeof messages.$inferSelect)[]>;
|
||||
createMessage(conversationId: number, role: string, content: string): Promise<typeof messages.$inferSelect>;
|
||||
}
|
||||
|
||||
export const chatStorage: IChatStorage = {
|
||||
async getConversation(id: number) {
|
||||
const [conversation] = await db.select().from(conversations).where(eq(conversations.id, id));
|
||||
return conversation;
|
||||
},
|
||||
|
||||
async getAllConversations() {
|
||||
return db.select().from(conversations).orderBy(desc(conversations.createdAt));
|
||||
},
|
||||
|
||||
async createConversation(title: string) {
|
||||
const [conversation] = await db.insert(conversations).values({ title }).returning();
|
||||
return conversation;
|
||||
},
|
||||
|
||||
async deleteConversation(id: number) {
|
||||
await db.delete(messages).where(eq(messages.conversationId, id));
|
||||
await db.delete(conversations).where(eq(conversations.id, id));
|
||||
},
|
||||
|
||||
async getMessagesByConversation(conversationId: number) {
|
||||
return db.select().from(messages).where(eq(messages.conversationId, conversationId)).orderBy(messages.createdAt);
|
||||
},
|
||||
|
||||
async createMessage(conversationId: number, role: string, content: string) {
|
||||
const [message] = await db.insert(messages).values({ conversationId, role, content }).returning();
|
||||
return message;
|
||||
},
|
||||
};
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
import fs from "node:fs";
|
||||
import OpenAI, { toFile } from "openai";
|
||||
import { Buffer } from "node:buffer";
|
||||
|
||||
export const openai = new OpenAI({
|
||||
apiKey: process.env.AI_INTEGRATIONS_OPENAI_API_KEY,
|
||||
baseURL: process.env.AI_INTEGRATIONS_OPENAI_BASE_URL,
|
||||
});
|
||||
|
||||
/**
|
||||
* Generate an image and return as Buffer.
|
||||
* Uses gpt-image-1 model via Replit AI Integrations.
|
||||
*/
|
||||
export async function generateImageBuffer(
|
||||
prompt: string,
|
||||
size: "1024x1024" | "512x512" | "256x256" = "1024x1024"
|
||||
): Promise<Buffer> {
|
||||
const response = await openai.images.generate({
|
||||
model: "gpt-image-1",
|
||||
prompt,
|
||||
size,
|
||||
});
|
||||
const base64 = response.data[0]?.b64_json ?? "";
|
||||
return Buffer.from(base64, "base64");
|
||||
}
|
||||
|
||||
/**
|
||||
* Edit/combine multiple images into a composite.
|
||||
* Uses gpt-image-1 model via Replit AI Integrations.
|
||||
*/
|
||||
export async function editImages(
|
||||
imageFiles: string[],
|
||||
prompt: string,
|
||||
outputPath?: string
|
||||
): Promise<Buffer> {
|
||||
const images = await Promise.all(
|
||||
imageFiles.map((file) =>
|
||||
toFile(fs.createReadStream(file), file, {
|
||||
type: "image/png",
|
||||
})
|
||||
)
|
||||
);
|
||||
|
||||
const response = await openai.images.edit({
|
||||
model: "gpt-image-1",
|
||||
image: images,
|
||||
prompt,
|
||||
});
|
||||
|
||||
const imageBase64 = response.data[0]?.b64_json ?? "";
|
||||
const imageBytes = Buffer.from(imageBase64, "base64");
|
||||
|
||||
if (outputPath) {
|
||||
fs.writeFileSync(outputPath, imageBytes);
|
||||
}
|
||||
|
||||
return imageBytes;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,3 @@
|
||||
export { registerImageRoutes } from "./routes";
|
||||
export { openai, generateImageBuffer, editImages } from "./client";
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
import type { Express, Request, Response } from "express";
|
||||
import { openai } from "./client";
|
||||
|
||||
export function registerImageRoutes(app: Express): void {
|
||||
app.post("/api/generate-image", async (req: Request, res: Response) => {
|
||||
try {
|
||||
const { prompt, size = "1024x1024" } = req.body;
|
||||
|
||||
if (!prompt) {
|
||||
return res.status(400).json({ error: "Prompt is required" });
|
||||
}
|
||||
|
||||
const response = await openai.images.generate({
|
||||
model: "gpt-image-1",
|
||||
prompt,
|
||||
n: 1,
|
||||
size: size as "1024x1024" | "512x512" | "256x256",
|
||||
});
|
||||
|
||||
const imageData = response.data[0];
|
||||
res.json({
|
||||
url: imageData.url,
|
||||
b64_json: imageData.b64_json,
|
||||
});
|
||||
} catch (error) {
|
||||
console.error("Error generating image:", error);
|
||||
res.status(500).json({ error: "Failed to generate image" });
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user