import { Queue, Worker } from "bullmq";
const searchQueue = new Queue("perplexity-research", {
connection: { host: "localhost", port: 6379 },
});
async function submitResearchBatch(
queries: string[],
callbackUrl: string,
model: string = "sonar-pro"
) {
const batchId = crypto.randomUUID();
for (const query of queries) {
await searchQueue.add("search", { batchId, query, callbackUrl, model }, {
attempts: 3,
backoff: { type: "exponential", delay: 2000 },
});
}
return { batchId, totalQueries: queries.length };
}
const worker = new Worker("perplexity-research", async (job) => {
const { query, callbackUrl, batchId, model } = job.data;
const response = await perplexity.chat.completions.create({
model,
messages: [{ role: "user", content: query }],
max_tokens: 2048,
});
const result = {
event: "perplexity.search.completed",
batchId,
query,
answer: response.choices[0].message.content,
citations: (response as any).citations || [],
model: response.model,
tokens: response.usage?.total_tokens,
};
await fetch(callbackUrl, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify(result),
});
}, {
connection: { host: "localhost", port: 6379 },
concurrency: 3,
limiter: { max: 40, duration: 60000 },
});
async function monitorTopics(
topics: string[],
webhookUrl: string
) {
for (const topic of topics) {
const response = await perplexity.chat.completions.create({
model: "sonar",
messages: [{
role: "system",
content: "Summarize the latest developments. Be concise. Include only new information.",
}, {
role: "user",
content: `Latest developments about "${topic}" in the past 24 hours`,
}],
search_recency_filter: "day",
max_tokens: 500,
} as any);
const answer = response.choices[0].message.content || "";
const citations = (response as any).citations || [];
if (citations.length > 0 && answer.length > 100) {
await fetch(webhookUrl, {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
event: "perplexity.monitor.update",
topic,
summary: answer,
citations,
timestamp: new Date().toISOString(),
}),
});
}
await new Promise((r) => setTimeout(r, 2000));
}
}
function consumeSearchStream(
query: string,
onText: (text: string) => void,
onCitations: (urls: string[]) => void,
onDone: () => void
) {
fetch("/api/search/stream", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({ query }),
}).then(async (response) => {
const reader = response.body!.getReader();
const decoder = new TextDecoder();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const lines = decoder.decode(value).split("\n");
for (const line of lines) {
if (!line.startsWith("data: ")) continue;
const event = JSON.parse(line.slice(6));
if (event.type === "text") onText(event.content);
if (event.type === "citations") onCitations(event.urls);
if (event.type === "done") onDone();
}
}
});
}
