server.js

import express from 'express';
// **CRITICAL: Import the 'fal' object directly for configuration**
import { fal } from '@fal-ai/client'; // Make sure 'fal' is the object you configure

// --- Key Management Setup ---
// Read comma-separated keys from the SINGLE environment variable FAL_KEY
const FAL_KEY_STRING = process.env.FAL_KEY;
const API_KEY = process.env.API_KEY; // Custom API Key for proxy auth

if (!FAL_KEY_STRING) {
    console.error("Error: FAL_KEY environment variable is not set.");
    console.error("Ensure FAL_KEY contains a comma-separated list of your Fal AI keys.");
    process.exit(1);
}

// Parse the comma-separated keys from FAL_KEY_STRING
const falKeys = FAL_KEY_STRING.split(',')
    .map(key => key.trim()) // Remove leading/trailing whitespace
    .filter(key => key.length > 0); // Remove any empty strings resulting from extra commas

if (falKeys.length === 0) {
    console.error("Error: No valid FAL keys found in the FAL_KEY environment variable after parsing.");
    console.error("Ensure FAL_KEY is a comma-separated list, e.g., 'key1,key2,key3'.");
    process.exit(1);
}

if (!API_KEY) {
    console.error("Error: API_KEY environment variable is not set.");
    process.exit(1);
}

let currentKeyIndex = 0;
// **NEW: Keep track of keys that failed persistently during runtime**
const invalidKeys = new Set();

console.log(`Loaded ${falKeys.length} Fal AI Key(s) from the FAL_KEY environment variable.`);

// **MODIFIED: Function to get the next *valid* key**
function getNextValidKey() {
    // Check if all keys have been marked as invalid
    if (invalidKeys.size >= falKeys.length) {
        console.error("All Fal AI keys are marked as invalid.");
        return null; // No valid keys left
    }

    const initialIndex = currentKeyIndex;
    let attempts = 0; // Prevent infinite loops in edge cases
    while (attempts < falKeys.length) {
        const keyIndex = currentKeyIndex % falKeys.length;
        const key = falKeys[keyIndex];

        // Move to the next index for the *next* call, regardless of validity
        currentKeyIndex = (keyIndex + 1) % falKeys.length;

        // Check if the current key is NOT in the invalid set
        if (!invalidKeys.has(key)) {
            // Found a valid key
            console.log(`Using Fal Key index: ${keyIndex} (from FAL_KEY list)`);
            return { key, index: keyIndex }; // Return the key and its original index
        } else {
             console.log(`Skipping invalid Fal Key index: ${keyIndex}`);
        }

        attempts++;
        // If the loop started at an invalid key and wrapped around,
        // check if we've checked all keys since the start
        if (currentKeyIndex === initialIndex && attempts > 0) {
            // This check might be redundant due to the invalidKeys.size check at the top,
            // but serves as an extra safeguard.
             console.warn("Looped through all keys, potentially all are invalid.");
            break;
        }
    }

    // If we exit the loop, it means no valid key was found
    console.error("Could not find a valid Fal AI key after checking all potentially available keys.");
    return null;
}

// **NEW/MODIFIED: Function to check if an error is likely related to a bad key**
// NOTE: This is a heuristic. You might need to adjust based on actual errors from Fal AI.
function isKeyRelatedError(error) {
    // Check if error itself is null/undefined
    if (!error) return false;

    const message = error.message?.toLowerCase() || '';
    // Check if the error object has a 'status' property (common in HTTP errors)
    const status = error.status || error.statusCode; // Check common status properties

    // Check for specific HTTP status codes indicative of auth/permission issues
    // 401: Unauthorized, 403: Forbidden, 429: Too Many Requests (often quota related)
    if (status === 401 || status === 403 || status === 429) {
        console.warn(`Detected potential key-related error (HTTP Status: ${status}).`);
        return true;
    }

    // Check for common error message patterns (case-insensitive)
    // Add more specific Fal AI error messages as you encounter them
    if (message.includes('invalid api key') ||
        message.includes('authentication failed') ||
        message.includes('permission denied') ||
        message.includes('quota exceeded') || // Treat quota errors as key-related for rotation
        message.includes('forbidden') ||
        message.includes('unauthorized') ||
        message.includes('rate limit') || // Often linked to key limits
        message.includes('credentials')) // Generic credential errors
        {
        console.warn(`Detected potential key-related error (message: "${message}")`);
        return true;
    }

    // Add more specific checks based on observed Fal AI errors if needed
    // Example: Check for specific error codes if Fal AI provides them
    // if (error.code === 'SOME_FAL_AUTH_ERROR_CODE') {
    //     return true;
    // }

    return false;
}
// --- End Key Management Setup ---

const app = express();
app.use(express.json({ limit: '50mb' }));
app.use(express.urlencoded({ extended: true, limit: '50mb' }));

const PORT = process.env.PORT || 3000;

// API Key 鉴权中间件 (unchanged)
const apiKeyAuth = (req, res, next) => {
    const authHeader = req.headers['authorization'];

    if (!authHeader) {
        console.warn('Unauthorized: No Authorization header provided');
        return res.status(401).json({ error: 'Unauthorized: No API Key provided' });
    }

    const authParts = authHeader.split(' ');
    if (authParts.length !== 2 || authParts[0].toLowerCase() !== 'bearer') {
        console.warn('Unauthorized: Invalid Authorization header format');
        return res.status(401).json({ error: 'Unauthorized: Invalid Authorization header format' });
    }

    const providedKey = authParts[1];
    if (providedKey !== API_KEY) {
        console.warn('Unauthorized: Invalid API Key');
        return res.status(401).json({ error: 'Unauthorized: Invalid API Key' });
    }

    next();
};

app.use(['/v1/models', '/v1/chat/completions'], apiKeyAuth);

// === 全局定义限制 === (unchanged)
const PROMPT_LIMIT = 4800;
const SYSTEM_PROMPT_LIMIT = 4800;
// === 限制定义结束 ===

// 定义 fal-ai/any-llm 支持的模型列表 (unchanged)
const FAL_SUPPORTED_MODELS = [ /* ... model list ... */ ];
// Helper function getOwner (unchanged)
const getOwner = (modelId) => { /* ... */ };
// GET /v1/models endpoint (unchanged)
app.get('/v1/models', (req, res) => { /* ... */ });
// convertMessagesToFalPrompt 函数 (unchanged)
function convertMessagesToFalPrompt(messages) { /* ... */ }
// === convertMessagesToFalPrompt 函数结束 ===


// --- NEW: Helper function to make Fal AI request with retries ---
async function makeFalRequestWithRetry(falInput, stream = false) {
    let attempts = 0;
    // Max attempts should be the total number of keys initially available
    const maxAttempts = falKeys.length;
    // Keep track of keys tried *within this specific request attempt*
    // This prevents infinite loops if getNextValidKey had issues,
    // and ensures we try each *available* key at most once per request.
    const attemptedKeysInThisRequest = new Set();

    while (attempts < maxAttempts) {
        const keyInfo = getNextValidKey(); // Get the next *valid* key info { key, index }

        if (!keyInfo) {
            // This should only happen if all keys are in the invalidKeys set
            console.error("makeFalRequestWithRetry: No valid Fal AI keys remaining.");
            throw new Error("No valid Fal AI keys available (all marked as invalid).");
        }

        // Prevent retrying the exact same key if getNextValidKey logic had an issue
        // or if a key wasn't marked invalid correctly on a previous attempt within this request.
        if (attemptedKeysInThisRequest.has(keyInfo.key)) {
            console.warn(`Key at index ${keyInfo.index} was already attempted for this request. Skipping to find next.`);
            // Don't increment 'attempts' here as we didn't use the key. Let the loop find the next.
            // If all keys end up being skipped, the `!keyInfo` check or `attempts < maxAttempts` will eventually handle it.
             continue;
        }
        attemptedKeysInThisRequest.add(keyInfo.key);
        attempts++; // Count this as a distinct attempt with a unique key for this request

        try {
            console.log(`Attempt ${attempts}/${maxAttempts}: Trying Fal Key index ${keyInfo.index}...`);

            // *** CRITICAL: Reconfigure the global fal client with the selected key for this attempt ***
            // Warning: This reconfigures the GLOBAL client. If you have many concurrent requests,
            // this could lead to race conditions. Consider instance isolation or a pool manager
            // for high-concurrency scenarios if the library doesn't support per-request credentials easily.
            console.warn(`Configuring GLOBAL fal client with key index ${keyInfo.index}. Ensure this is safe for your concurrency model.`);
            fal.config({ credentials: keyInfo.key }); // Use the specific key for this attempt

            if (stream) {
                // Use the configured global 'fal' object for the stream request
                const falStream = await fal.stream("fal-ai/any-llm", { input: falInput });
                console.log(`Successfully initiated stream with key index ${keyInfo.index}.`);
                // Success! Return the stream iterator directly
                return falStream;
            } else {
                // Use the configured global 'fal' object for the non-stream request
                console.log(`Executing non-stream request with key index ${keyInfo.index}...`);
                // Assuming fal.subscribe or similar method for non-streaming
                // Adapt this line if your non-stream method is different
                const result = await fal.subscribe("fal-ai/any-llm", { input: falInput, logs: true });
                console.log(`Successfully received non-stream result with key index ${keyInfo.index}.`);

                // Check for errors *within* the successful response structure (if applicable)
                if (result && result.error) {
                     console.error(`Fal-ai returned an error in the result payload (Key Index ${keyInfo.index}):`, result.error);
                     // Decide if this specific payload error should also invalidate the key
                     if (isKeyRelatedError(result.error)) { // Reuse the checker
                         console.warn(`Marking Fal Key index ${keyInfo.index} as invalid due to error in response payload.`);
                         invalidKeys.add(keyInfo.key);
                         // Continue the loop to try the next key
                         continue; // Go to the next iteration of the while loop
                     } else {
                         // Throw an error that will be caught by the outer handler, not retried
                         throw new Error(`Fal-ai error in result payload: ${JSON.stringify(result.error)}`);
                     }
                 }
                 // Success! Return the result
                return result;
            }
        } catch (error) {
            console.error(`Error using Fal Key index ${keyInfo.index}:`, error.message || error);

            // Check if the caught error indicates the key is invalid
            if (isKeyRelatedError(error)) {
                console.warn(`Marking Fal Key index ${keyInfo.index} as invalid due to caught error.`);
                // **ACTION: Add the failed key to the set of invalid keys**
                invalidKeys.add(keyInfo.key);
                // Continue to the next iteration of the while loop to try another key
            } else {
                // Error is not key-related (e.g., network issue, bad input, internal Fal error)
                // Don't retry with other keys for this type of error. Fail the request immediately.
                console.error("Error does not appear to be key-related. Failing request without further key retries.");
                throw error; // Re-throw the original error to be caught by the main endpoint handler
            }
        }
    } // End while loop

    // If the loop finishes, it means all available keys were tried and failed with key-related errors
    throw new Error(`Request failed after trying ${attempts} unique Fal key(s). All failed with key-related errors or were already marked invalid.`);
}


// POST /v1/chat/completions endpoint (MODIFIED to use retry logic)
app.post('/v1/chat/completions', async (req, res) => {
    const { model, messages, stream = false, reasoning = false, /* ...restOpenAIParams */ } = req.body;

    console.log(`--> POST /v1/chat/completions | Model: ${model} | Stream: ${stream}`);

    // Basic Validation (unchanged)
    if (!FAL_SUPPORTED_MODELS.includes(model)) {
        console.warn(`Warning: Requested model '${model}' is not in the explicitly supported list. Proxy will still attempt.`);
    }
    if (!model || !messages || !Array.isArray(messages) || messages.length === 0) {
        console.error("Invalid request: Missing 'model' or 'messages' array.");
        return res.status(400).json({ error: 'Missing or invalid parameters: model and messages array are required.' });
    }

    try {
        // --- Prepare Input (unchanged) ---
        const { prompt, system_prompt } = convertMessagesToFalPrompt(messages);
        const falInput = {
            model: model,
            prompt: prompt,
            ...(system_prompt && { system_prompt: system_prompt }),
            reasoning: !!reasoning,
        };
        console.log("Attempting Fal request with key rotation/retry...");

        // --- Handle Stream vs Non-Stream using the retry helper ---
        if (stream) {
            res.setHeader('Content-Type', 'text/event-stream; charset=utf-8');
            res.setHeader('Cache-Control', 'no-cache');
            res.setHeader('Connection', 'keep-alive');
            res.setHeader('Access-Control-Allow-Origin', '*');
            res.flushHeaders();

            let previousOutput = '';
            let falStream; // Declare falStream here

            try {
                // **MODIFIED: Initiate stream using the retry logic helper**
                falStream = await makeFalRequestWithRetry(falInput, true);

                // Process the stream events (logic mostly unchanged)
                for await (const event of falStream) {
                    const currentOutput = (event && typeof event.output === 'string') ? event.output : '';
                    const isPartial = (event && typeof event.partial === 'boolean') ? event.partial : true;
                    const errorInfo = (event && event.error) ? event.error : null;

                    // Handle errors *within* the stream payload
                    if (errorInfo) {
                        console.error("Error received *within* fal stream event:", errorInfo);
                        // Optionally send an error chunk (check if needed)
                        const errorChunk = { /* ... error chunk details ... */ };
                        if (!res.writableEnded) {
                             res.write(`data: ${JSON.stringify(errorChunk)}\n\n`);
                        } else { console.warn("Stream ended before writing event error."); }
                         // Decide if this stream error should also invalidate the key
                         // Note: The key might have already been marked invalid if the error happened during initial connection
                         // You might need more context to decide if an *in-stream* error means the key is bad.
                         // For now, we just log it. If it causes the stream to terminate prematurely,
                         // the outer catch might handle it, but key invalidation might not occur unless the
                         // error object passed to the catch block triggers isKeyRelatedError.
                    }

                    // Calculate delta (logic unchanged)
                    let deltaContent = '';
                    if (currentOutput.startsWith(previousOutput)) {
                         deltaContent = currentOutput.substring(previousOutput.length);
                    } else if (currentOutput.length > 0) {
                         deltaContent = currentOutput;
                         previousOutput = '';
                    }
                    previousOutput = currentOutput;

                    // Send OpenAI compatible chunk (logic unchanged)
                    if (deltaContent || !isPartial) {
                        const openAIChunk = {
                            id: `chatcmpl-${Date.now()}`,
                            object: "chat.completion.chunk",
                            created: Math.floor(Date.now() / 1000),
                            model: model,
                            choices: [{ index: 0, delta: { content: deltaContent }, finish_reason: isPartial === false ? "stop" : null }]
                        };
                        if (!res.writableEnded) {
                            res.write(`data: ${JSON.stringify(openAIChunk)}\n\n`);
                        } else { console.warn("Stream ended before writing data chunk."); }
                    }
                } // End for-await loop

                // Send the final [DONE] marker (logic unchanged)
                if (!res.writableEnded) {
                    res.write(`data: [DONE]\n\n`);
                    res.end();
                    console.log("<-- Stream finished successfully.");
                } else {
                    console.log("<-- Stream finished, but connection was already ended.");
                }

            } catch (streamError) {
                // **MODIFIED: Catches errors from makeFalRequestWithRetry OR stream iteration**
                // This error could be a non-key-related error, or the "all keys failed" error.
                console.error('Error during stream request processing:', streamError.message || streamError);
                 try {
                     if (!res.headersSent) {
                         // Error likely occurred in makeFalRequestWithRetry before stream started
                         const errorMessage = (streamError instanceof Error) ? streamError.message : JSON.stringify(streamError);
                         res.status(502).json({ // 502 Bad Gateway is appropriate for upstream failure
                            error: 'Failed to initiate Fal stream',
                            details: errorMessage
                         });
                         console.log("<-- Stream initiation failed response sent.");
                     } else if (!res.writableEnded) {
                         // Stream started but failed during processing
                         const errorDetails = (streamError instanceof Error) ? streamError.message : JSON.stringify(streamError);
                         res.write(`data: ${JSON.stringify({ error: { message: "Stream processing error after initiation", type: "proxy_error", details: errorDetails } })}\n\n`);
                         res.write(`data: [DONE]\n\n`); // Still send DONE for client handling
                         res.end();
                         console.log("<-- Stream error sent, stream ended.");
                     } else {
                        console.log("<-- Stream error occurred, but connection already ended.");
                     }
                 } catch (finalError) {
                    console.error('Error sending stream error message to client:', finalError);
                    if (!res.writableEnded) { res.end(); }
                 }
            }

        } else {
            // --- Non-Stream ---
            try {
                // **MODIFIED: Get the result using the retry logic helper**
                const result = await makeFalRequestWithRetry(falInput, false);
                // console.log("Received non-stream result via retry function:", JSON.stringify(result, null, 2));

                // Construct OpenAI compatible response (logic unchanged)
                const openAIResponse = {
                    id: `chatcmpl-${result.requestId || Date.now()}`,
                    object: "chat.completion",
                    created: Math.floor(Date.now() / 1000),
                    model: model,
                    choices: [{ index: 0, message: { role: "assistant", content: result.output || "" }, finish_reason: "stop" }],
                    usage: { prompt_tokens: null, completion_tokens: null, total_tokens: null },
                    system_fingerprint: null,
                     ...(result.reasoning && { fal_reasoning: result.reasoning }),
                };

                res.json(openAIResponse);
                console.log("<-- Non-stream response sent successfully.");

            } catch (error) {
                 // **MODIFIED: Catches errors from makeFalRequestWithRetry**
                 // This error could be a non-key-related error, or the "all keys failed" error.
                console.error('Error during non-stream request processing:', error.message || error);
                if (!res.headersSent) {
                    const errorMessage = (error instanceof Error) ? error.message : JSON.stringify(error);
                    // Check if it was the "all keys failed" error
                    const finalMessage = errorMessage.includes("No valid Fal AI keys available") || errorMessage.includes("Request failed after trying")
                        ? `Fal request failed: ${errorMessage}` // Simplified message
                        : `Internal Server Error processing Fal request: ${errorMessage}`;
                    res.status(502).json({ error: 'Fal Request Failed', details: finalMessage }); // 502 Bad Gateway
                    console.log("<-- Non-stream error response sent.");
                } else {
                    console.error("Headers already sent for non-stream error? This is unexpected.");
                    if (!res.writableEnded) { res.end(); }
                }
            }
        }

    } catch (error) {
        // Catch errors from parameter validation or prompt conversion *before* calling Fal
        console.error('Unhandled error before initiating Fal request:', error.message || error);
        if (!res.headersSent) {
            const errorMessage = (error instanceof Error) ? error.message : JSON.stringify(error);
            res.status(500).json({ error: 'Internal Server Error in Proxy Setup', details: errorMessage });
            console.log("<-- Proxy setup error response sent.");
        } else {
             console.error("Headers already sent when catching setup error. Ending response.");
             if (!res.writableEnded) { res.end(); }
        }
    }
});

// 启动服务器 (Updated startup message)
app.listen(PORT, () => {
    console.log(`=====================================================================`);
    console.log(` Fal OpenAI Proxy Server (Multi-Key Rotation & Failover)`);
    console.log(`---------------------------------------------------------------------`);
    console.log(` Listening on port : ${PORT}`);
    console.log(` Reading Fal Keys from : FAL_KEY environment variable (comma-separated)`);
    console.log(` Loaded Keys Count   : ${falKeys.length}`);
    console.log(` Invalid Keys Set    : Initialized (size: ${invalidKeys.size})`); // Show invalid set size
    console.log(` API Key Auth        : ${API_KEY ? 'Enabled (using API_KEY env var)' : 'Disabled'}`);
    console.log(` Input Limits        : System Prompt=${SYSTEM_PROMPT_LIMIT}, Prompt=${PROMPT_LIMIT}`);
    console.log(` Concurrency Warning : Global Fal client reconfigured per request attempt!`);
    console.log(`---------------------------------------------------------------------`);
    console.log(` Endpoints:`);
    console.log(`   POST http://localhost:${PORT}/v1/chat/completions`);
    console.log(`   GET  http://localhost:${PORT}/v1/models`);
    console.log(`=====================================================================`);
});

// 根路径响应 (Updated message)
app.get('/', (req, res) => {
    res.send(`Fal OpenAI Proxy (Multi-Key Rotation from FAL_KEY) is running. Loaded ${falKeys.length} key(s). Currently ${invalidKeys.size} key(s) marked as invalid.`); // Show invalid count
});