initial commit: rocket.new export of broadcastbeat
This commit is contained in:
Ryan Salazar
238
src/app/api/ai/console/route.ts
Normal file
238
src/app/api/ai/console/route.ts
Normal file
@@ -0,0 +1,238 @@
|
||||
import { NextRequest, NextResponse } from 'next/server';
|
||||
import { createClient } from '@/lib/supabase/server';
|
||||
import { hybridAIStream, pingOllama } from '@/lib/ai/hybridRouter';
|
||||
|
||||
const ANTHROPIC_PRIMARY_MODEL = 'claude-sonnet-4-20250514';
|
||||
const ANTHROPIC_FALLBACK_MODEL = 'claude-3-5-sonnet-20241022';
|
||||
|
||||
// ─── Anthropic streaming helper ───────────────────────────────────────────────
|
||||
async function streamAnthropic(messages: any[], maxTokens: number): Promise<Response> {
|
||||
const apiKey = process.env.ANTHROPIC_API_KEY || '';
|
||||
const systemMsg = messages.find((m: any) => m.role === 'system');
|
||||
const userMessages = messages.filter((m: any) => m.role !== 'system');
|
||||
|
||||
const tryModel = async (model: string): Promise<Response> => {
|
||||
const payload: any = {
|
||||
model,
|
||||
max_tokens: maxTokens,
|
||||
stream: true,
|
||||
messages: userMessages,
|
||||
};
|
||||
if (systemMsg) payload.system = systemMsg.content;
|
||||
|
||||
return fetch('https://api.anthropic.com/v1/messages', {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'Content-Type': 'application/json',
|
||||
'x-api-key': apiKey,
|
||||
'anthropic-version': '2023-06-01',
|
||||
},
|
||||
body: JSON.stringify(payload),
|
||||
});
|
||||
};
|
||||
|
||||
let response = await tryModel(ANTHROPIC_PRIMARY_MODEL);
|
||||
if (!response.ok && response.status === 404) {
|
||||
response = await tryModel(ANTHROPIC_FALLBACK_MODEL);
|
||||
}
|
||||
return response;
|
||||
}
|
||||
|
||||
// ─── Convert Anthropic SSE stream to our SSE format ──────────────────────────
|
||||
function anthropicStreamToSSE(
|
||||
anthropicStream: ReadableStream,
|
||||
encoder: TextEncoder
|
||||
): ReadableStream {
|
||||
const reader = anthropicStream.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = '';
|
||||
|
||||
return new ReadableStream({
|
||||
async start(controller) {
|
||||
try {
|
||||
controller.enqueue(encoder.encode(`data: ${JSON.stringify({ type: 'start', provider: 'anthropic' })}\n\n`));
|
||||
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
const lines = buffer.split('\n');
|
||||
buffer = lines.pop() || '';
|
||||
|
||||
for (const line of lines) {
|
||||
if (line.startsWith('data: ')) {
|
||||
try {
|
||||
const data = JSON.parse(line.slice(6));
|
||||
if (data.type === 'content_block_delta' && data.delta?.text) {
|
||||
controller.enqueue(encoder.encode(`data: ${JSON.stringify({ type: 'chunk', chunk: data.delta.text })}\n\n`));
|
||||
} else if (data.type === 'message_stop') {
|
||||
controller.enqueue(encoder.encode(`data: ${JSON.stringify({ type: 'done' })}\n\n`));
|
||||
}
|
||||
} catch { /* skip invalid JSON */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
controller.enqueue(encoder.encode(`data: ${JSON.stringify({ type: 'done' })}\n\n`));
|
||||
controller.close();
|
||||
} catch (err: any) {
|
||||
controller.enqueue(encoder.encode(`data: ${JSON.stringify({ type: 'error', error: err.message })}\n\n`));
|
||||
controller.close();
|
||||
}
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
// ─── Convert Ollama SSE stream to our SSE format ─────────────────────────────
|
||||
function ollamaStreamToSSE(
|
||||
ollamaStream: ReadableStream,
|
||||
encoder: TextEncoder
|
||||
): ReadableStream {
|
||||
const reader = ollamaStream.getReader();
|
||||
const decoder = new TextDecoder();
|
||||
let buffer = '';
|
||||
|
||||
return new ReadableStream({
|
||||
async start(controller) {
|
||||
try {
|
||||
controller.enqueue(encoder.encode(`data: ${JSON.stringify({ type: 'start', provider: 'ollama' })}\n\n`));
|
||||
|
||||
while (true) {
|
||||
const { done, value } = await reader.read();
|
||||
if (done) break;
|
||||
|
||||
buffer += decoder.decode(value, { stream: true });
|
||||
const lines = buffer.split('\n');
|
||||
buffer = lines.pop() || '';
|
||||
|
||||
for (const line of lines) {
|
||||
if (line.startsWith('data: ')) {
|
||||
try {
|
||||
const data = JSON.parse(line.slice(6));
|
||||
// OpenAI-compatible format from Ollama
|
||||
const chunk = data?.choices?.[0]?.delta?.content;
|
||||
if (chunk) {
|
||||
controller.enqueue(encoder.encode(`data: ${JSON.stringify({ type: 'chunk', chunk })}\n\n`));
|
||||
}
|
||||
if (data?.choices?.[0]?.finish_reason === 'stop') {
|
||||
controller.enqueue(encoder.encode(`data: ${JSON.stringify({ type: 'done' })}\n\n`));
|
||||
}
|
||||
} catch { /* skip */ }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
controller.enqueue(encoder.encode(`data: ${JSON.stringify({ type: 'done' })}\n\n`));
|
||||
controller.close();
|
||||
} catch (err: any) {
|
||||
controller.enqueue(encoder.encode(`data: ${JSON.stringify({ type: 'error', error: err.message })}\n\n`));
|
||||
controller.close();
|
||||
}
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
export async function POST(request: NextRequest) {
|
||||
try {
|
||||
const supabase = await createClient();
|
||||
const { data: { user }, error: authError } = await supabase.auth.getUser();
|
||||
if (authError || !user) return NextResponse.json({ error: 'Unauthorized' }, { status: 401 });
|
||||
|
||||
const { data: profile } = await supabase.from('user_profiles').select('role').eq('id', user.id).single();
|
||||
if (!['administrator', 'admin'].includes(profile?.role || '')) {
|
||||
return NextResponse.json({ error: 'Forbidden — AI Console is admin only' }, { status: 403 });
|
||||
}
|
||||
|
||||
const body = await request.json();
|
||||
const { messages, stream = true, maxTokens = 2000, sessionId } = body;
|
||||
|
||||
if (!messages?.length) {
|
||||
return NextResponse.json({ error: 'messages required' }, { status: 400 });
|
||||
}
|
||||
|
||||
const encoder = new TextEncoder();
|
||||
|
||||
if (stream) {
|
||||
let provider: 'ollama' | 'anthropic' = 'anthropic';
|
||||
let model = ANTHROPIC_PRIMARY_MODEL;
|
||||
let readable: ReadableStream;
|
||||
|
||||
try {
|
||||
const result = await hybridAIStream(messages, { maxTokens, priority: 1 });
|
||||
provider = result.provider;
|
||||
model = result.model;
|
||||
|
||||
if (provider === 'ollama') {
|
||||
readable = ollamaStreamToSSE(result.stream, encoder);
|
||||
} else {
|
||||
readable = anthropicStreamToSSE(result.stream, encoder);
|
||||
}
|
||||
} catch {
|
||||
// Last resort: direct Anthropic stream
|
||||
provider = 'anthropic';
|
||||
const anthropicResponse = await streamAnthropic(messages, maxTokens);
|
||||
if (!anthropicResponse.ok || !anthropicResponse.body) {
|
||||
return NextResponse.json({ error: 'All AI providers unavailable' }, { status: 503 });
|
||||
}
|
||||
readable = anthropicStreamToSSE(anthropicResponse.body, encoder);
|
||||
}
|
||||
|
||||
// Save session message (fire-and-forget)
|
||||
if (sessionId) {
|
||||
supabase.from('ai_console_sessions')
|
||||
.update({ last_message_at: new Date().toISOString(), provider_used: provider, model_used: model })
|
||||
.eq('id', sessionId)
|
||||
.then(() => {});
|
||||
}
|
||||
|
||||
return new NextResponse(readable, {
|
||||
headers: {
|
||||
'Content-Type': 'text/event-stream',
|
||||
'Cache-Control': 'no-cache',
|
||||
'Connection': 'keep-alive',
|
||||
'X-AI-Provider': provider,
|
||||
'X-AI-Model': model,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
// Non-streaming (for background tasks)
|
||||
const { hybridAI } = await import('@/lib/ai/hybridRouter');
|
||||
const result = await hybridAI(messages, { maxTokens, priority: 1 });
|
||||
return NextResponse.json({
|
||||
text: result.text,
|
||||
provider: result.provider,
|
||||
model: result.model,
|
||||
latencyMs: result.latencyMs,
|
||||
});
|
||||
} catch (err: any) {
|
||||
return NextResponse.json({ error: err.message }, { status: 500 });
|
||||
}
|
||||
}
|
||||
|
||||
// ─── GET — ping / status ──────────────────────────────────────────────────────
|
||||
export async function GET(request: NextRequest) {
|
||||
try {
|
||||
const supabase = await createClient();
|
||||
const { data: { user } } = await supabase.auth.getUser();
|
||||
if (!user) return NextResponse.json({ error: 'Unauthorized' }, { status: 401 });
|
||||
|
||||
const { searchParams } = new URL(request.url);
|
||||
const action = searchParams.get('action');
|
||||
|
||||
if (action === 'ping') {
|
||||
const ollamaStatus = await pingOllama();
|
||||
const anthropicKey = process.env.ANTHROPIC_API_KEY;
|
||||
return NextResponse.json({
|
||||
ollama: ollamaStatus,
|
||||
anthropic: { configured: !!anthropicKey },
|
||||
currentProvider: ollamaStatus.online ? 'ollama' : 'anthropic',
|
||||
});
|
||||
}
|
||||
|
||||
return NextResponse.json({ error: 'Unknown action' }, { status: 400 });
|
||||
} catch (err: any) {
|
||||
return NextResponse.json({ error: err.message }, { status: 500 });
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user