Frederik Spirgi
204 lines
14 KiB
JavaScript
204 lines
14 KiB
JavaScript
import sharp from "sharp";
|
|
import {createHash, randomUUID} from "node:crypto";
|
|
import {attachmentMetadata} from "../db/attachments.js";
|
|
import {LocalFileStorage} from "./localFileStorage.js";
|
|
import {VISION_PROMPT_VERSION, VISION_RESPONSE_SCHEMA, VISION_ANALYSIS_PROMPT, VISION_EVIDENCE_GUIDANCE, normalizeVisionQuestion, normalizeVisionAnalysis} from "./visionAnalysis.js";
|
|
|
|
export class AttachmentError extends Error {
|
|
constructor(message, status = 400) { super(message); this.status = status; }
|
|
}
|
|
export function parseVisionAnalysis(text) {
|
|
const raw = String(text || "").trim();
|
|
if (!raw) throw new AttachmentError("Image analysis returned no information. Please retry.", 502);
|
|
try {
|
|
// Ollama constrains output with a JSON schema; tolerate a whole fenced
|
|
// response from older/proxy backends, never extract a guessed JSON fragment.
|
|
const json = raw.startsWith("```json\n") && raw.endsWith("```") ? raw.slice(8, -3).trim()
|
|
: raw.startsWith("```\n") && raw.endsWith("```") ? raw.slice(4, -3).trim() : raw;
|
|
const value = JSON.parse(json);
|
|
if (!value || typeof value.summary !== "string") throw new Error("Invalid analysis");
|
|
return normalizeVisionAnalysis(value);
|
|
} catch {
|
|
return {...normalizeVisionAnalysis({summary: raw}), rawAnalysis: raw.slice(0, 6000)};
|
|
}
|
|
}
|
|
|
|
export function createAttachmentService(db, config, modelService) {
|
|
const storage = new LocalFileStorage(config.uploadDir);
|
|
const log = (event, fields) => { if (config.structuredRequestLoggingEnabled) console.log(JSON.stringify({event, ...fields})); };
|
|
const get = id => db.prepare("SELECT * FROM attachments WHERE id = ?").get(id);
|
|
const rows = messageId => db.prepare("SELECT a.*, m.position FROM attachments a JOIN message_attachments m ON m.attachment_id=a.id WHERE m.message_id=? ORDER BY m.position").all(messageId);
|
|
const refersToEarlierImage = question => {
|
|
const text = String(question || "");
|
|
return /\b(?:previous|prior|earlier|first|last)\s+(?:(?:attached|uploaded)\s+)?(?:image|photo|picture|screenshot)\b|\b(?:image|photo|picture|screenshot)\s+(?:from\s+)?(?:before|earlier|previously)\b/i.test(text)
|
|
|| (/\b(?:image|photo|picture|screenshot)\b/i.test(text) && /\b(?:previous|prior|earlier|first|last)\s+one\b/i.test(text));
|
|
};
|
|
const selectImages = (messages, question) => {
|
|
const currentMessage = [...messages].reverse().find(message => message.role === "user");
|
|
const currentMessageImages = currentMessage?.id != null ? rows(currentMessage.id).map((row, index) => ({
|
|
...row, messageId: currentMessage.id, scope: "current", label: `Current image ${index + 1}`
|
|
})) : [];
|
|
const historicalImages = [];
|
|
if (refersToEarlierImage(question) && (!currentMessageImages.length || /\b(?:previous|prior|earlier)\b/i.test(String(question || "")))) {
|
|
const earlier = messages.filter(message => message.id !== currentMessage?.id);
|
|
const candidates = earlier.flatMap(message => rows(message.id).map(row => ({...row, messageId: message.id})));
|
|
const distinctCandidates = candidates.filter(row => !currentMessageImages.some(image => image.id === row.id));
|
|
const row = /\bfirst\b/i.test(question) ? distinctCandidates[0] : distinctCandidates.at(-1);
|
|
if (row) {
|
|
historicalImages.push({...row, scope: "historical", label: `Earlier image from message ${row.messageId} (image ${row.position + 1})`});
|
|
}
|
|
}
|
|
return {currentMessageImages, historicalImages};
|
|
};
|
|
const selectedRows = selection => [...selection.currentMessageImages, ...selection.historicalImages];
|
|
const cached = (row, question) => {
|
|
const entry = db.prepare("SELECT analysis FROM attachment_vision_analyses WHERE attachment_id=? AND vision_model=? AND prompt_version=? AND question=?")
|
|
.get(row.id, config.visionModel, VISION_PROMPT_VERSION, normalizeVisionQuestion(question));
|
|
if (!entry) return null;
|
|
try { const analysis = JSON.parse(entry.analysis); return typeof analysis?.summary === "string" ? analysis : null; }
|
|
catch { return null; }
|
|
};
|
|
const analyzeRows = async (images, question, signal, diagnostics = {}) => {
|
|
for (const row of images) {
|
|
signal?.throwIfAborted();
|
|
const fields = {requestId: diagnostics.requestId ?? null, chatId: diagnostics.chatId ?? null,
|
|
triggeringMessageId: diagnostics.triggeringMessageId ?? row.messageId, attachmentId: row.id,
|
|
sourceMessageId: row.messageId, attachmentPosition: row.position, scope: row.scope || "current",
|
|
storageKey: row.storage_key, originalFilename: row.original_name, visionModel: config.visionModel,
|
|
visionPromptVersion: VISION_PROMPT_VERSION, question: normalizeVisionQuestion(question).slice(0, 500)};
|
|
const hit = cached(row, question);
|
|
if (hit) { log("visionAnalysisCacheHit", {...fields, cacheHit: true, resultSummary: hit.summary.slice(0, 160)}); continue; }
|
|
try {
|
|
log("visionAnalysisStart", {...fields, cacheHit: false});
|
|
const buffer = await storage.read(row.storage_key);
|
|
if (config.visionDebugSha256Enabled && config.structuredRequestLoggingEnabled) {
|
|
log("visionImageRead", {...fields, sha256: createHash("sha256").update(buffer).digest("hex")});
|
|
}
|
|
const result = await modelService.generateReply(config.visionModel, [
|
|
{role: "system", content: VISION_ANALYSIS_PROMPT},
|
|
{role: "user", content: `User question: ${JSON.stringify(String(question ?? ""))}\nAnalyze this image specifically to help answer that question.`, images: [buffer.toString("base64")]}
|
|
], {signal, reasoning: false, format: VISION_RESPONSE_SCHEMA, generation: {temperature: 0.1, responseLength: 1600}});
|
|
const parsed = parseVisionAnalysis(result);
|
|
const analysis = JSON.stringify(parsed);
|
|
signal?.throwIfAborted();
|
|
db.transaction(() => {
|
|
db.prepare(`INSERT INTO attachment_vision_analyses(attachment_id,vision_model,prompt_version,question,analysis)
|
|
VALUES(?,?,?,?,?) ON CONFLICT(attachment_id,vision_model,prompt_version,question)
|
|
DO UPDATE SET analysis=excluded.analysis,analyzed_at=CURRENT_TIMESTAMP`)
|
|
.run(row.id, config.visionModel, VISION_PROMPT_VERSION, normalizeVisionQuestion(question), analysis);
|
|
// Keep legacy snapshot fields for compatibility; cache reads use
|
|
// only the versioned, question-specific records above.
|
|
db.prepare("UPDATE attachments SET vision_model=?,vision_analysis=?,vision_analyzed_at=CURRENT_TIMESTAMP WHERE id=?")
|
|
.run(config.visionModel, analysis, row.id);
|
|
})();
|
|
log("visionAnalysisComplete", {...fields, cacheHit: false, resultSummary: parsed.summary.slice(0, 160)});
|
|
} catch (error) {
|
|
if (signal?.aborted) throw error;
|
|
throw new AttachmentError("Image analysis failed. Check that the vision model is available, then retry.", 502);
|
|
}
|
|
}
|
|
};
|
|
let cleaning = null;
|
|
let closed = false;
|
|
const cleanup = () => {
|
|
if (closed) return Promise.resolve();
|
|
if (cleaning) return cleaning;
|
|
cleaning = (async () => {
|
|
db.prepare("DELETE FROM attachments WHERE status='temporary' AND created_at < datetime('now','-24 hours')").run();
|
|
for (const row of db.prepare("SELECT storage_key FROM upload_deletions").all()) {
|
|
try {
|
|
await storage.delete(row.storage_key);
|
|
db.prepare("DELETE FROM upload_deletions WHERE storage_key=?").run(row.storage_key);
|
|
} catch { console.warn("Attachment file cleanup failed; will retry."); }
|
|
}
|
|
})().finally(() => { cleaning = null; });
|
|
return cleaning;
|
|
};
|
|
const timer = setInterval(() => { void cleanup(); }, 3600000);
|
|
timer.unref();
|
|
void cleanup();
|
|
return {
|
|
storage, cleanup,
|
|
async close() { clearInterval(timer); await cleanup(); closed = true; },
|
|
list(messageId) { return rows(messageId).map(attachmentMetadata); },
|
|
owned(id, user) {
|
|
const row = typeof id === "string" && /^[a-f0-9-]{36}$/.test(id) ? get(id) : null;
|
|
if (!row || row.username !== user) throw new AttachmentError("Attachment not found", 404);
|
|
if (row.status === "attached" && !db.prepare("SELECT 1 FROM message_attachments m JOIN chat_messages c ON c.id=m.message_id JOIN chat_sessions s ON s.id=c.chat_id WHERE m.attachment_id=? AND s.username=?").get(id, user)) throw new AttachmentError("Attachment not found", 404);
|
|
return row;
|
|
},
|
|
remove(id, user) {
|
|
const row = this.owned(id, user);
|
|
if (row.status !== "temporary") throw new AttachmentError("Attached images cannot be removed from the composer");
|
|
db.prepare("DELETE FROM attachments WHERE id=?").run(id);
|
|
return cleanup();
|
|
},
|
|
async upload(file, user) {
|
|
await cleanup();
|
|
if (!file?.buffer?.length) throw new AttachmentError("Choose a non-empty PNG, JPEG or WebP image");
|
|
if (file.size > config.uploadMaxSizeMb * 1024 * 1024) throw new AttachmentError("Image is too large", 413);
|
|
let info;
|
|
try {
|
|
const image = sharp(file.buffer, {limitInputPixels: config.uploadMaxPixels, failOn: "warning"});
|
|
info = await image.metadata();
|
|
if (!["png", "jpeg", "webp"].includes(info.format) || (info.pages || 1) > 1) throw new Error("Unsupported");
|
|
await image.stats(); // Decode pixels, not just headers: reject truncated/malformed images.
|
|
} catch { throw new AttachmentError("Invalid image. Use a non-animated PNG, JPEG or WebP within the pixel limit."); }
|
|
const {storageKey} = await storage.save({buffer: file.buffer, extension: info.format}).catch(() => { throw new AttachmentError("Image could not be saved. Please retry.", 503); });
|
|
const id = randomUUID();
|
|
try {
|
|
// Bound abandoned uploads per account as well as per submitted message.
|
|
db.transaction(() => {
|
|
if (db.prepare("SELECT COUNT(*) AS n FROM attachments WHERE username=? AND status='temporary'").get(user).n >= config.uploadMaxFiles * 4) throw new AttachmentError("Too many pending images. Remove unused uploads first.");
|
|
db.prepare("INSERT INTO attachments(id,username,original_name,storage_key,mime_type,size_bytes,width,height) VALUES(?,?,?,?,?,?,?,?)").run(id, user,
|
|
String(file.originalname || "image").replace(/[\\/\x00-\x1f]/g, "_").slice(0, 255), storageKey, `image/${info.format}`, file.size, info.width, info.height);
|
|
})();
|
|
} catch (error) { await storage.delete(storageKey); throw error; }
|
|
log("attachmentUploadSuccess", {id, size: file.size, mime: `image/${info.format}`});
|
|
return attachmentMetadata(get(id));
|
|
},
|
|
createMessage(ids, user, chatId, insert, reuseChatId = null) {
|
|
return db.transaction(() => {
|
|
if (!Array.isArray(ids) || ids.length > config.uploadMaxFiles || new Set(ids).size !== ids.length) throw new AttachmentError(`Use at most ${config.uploadMaxFiles} distinct images`);
|
|
for (const id of ids) {
|
|
const row = this.owned(id, user);
|
|
if (row.status !== "temporary" && !db.prepare("SELECT 1 FROM message_attachments m JOIN chat_messages c ON c.id=m.message_id WHERE m.attachment_id=? AND c.chat_id IN (?,?)").get(id, chatId, reuseChatId)) throw new AttachmentError("Image belongs to a different conversation");
|
|
}
|
|
const result = insert();
|
|
ids.forEach((id, index) => {
|
|
db.prepare("INSERT INTO message_attachments(message_id,attachment_id,position) VALUES(?,?,?)").run(Number(result.lastInsertRowid), id, index);
|
|
db.prepare("UPDATE attachments SET status='attached' WHERE id=?").run(id);
|
|
});
|
|
return result;
|
|
})();
|
|
},
|
|
copy(sourceId, targetId) { db.prepare("INSERT INTO message_attachments SELECT ?,attachment_id,position FROM message_attachments WHERE message_id=?").run(targetId, sourceId); },
|
|
selectImages,
|
|
searchContext(selection, question) {
|
|
const searchImages = selection.currentMessageImages.length ? selection.currentMessageImages : selection.historicalImages;
|
|
return searchImages.map(row => {
|
|
const analysis = cached(row, question);
|
|
if (!analysis) return "";
|
|
return [analysis.answer, ...analysis.visibleText, analysis.summary].filter(Boolean).join(" ").slice(0, 300);
|
|
}).join(" ").slice(0, 400);
|
|
},
|
|
hasImages: selection => selectedRows(selection).length > 0,
|
|
analyze: (messageId, question, signal) => analyzeRows(rows(messageId).map((row, index) => ({...row, messageId, scope: "current", label: `Current image ${index + 1}`})), question, signal),
|
|
analyzeContext: (selection, question, signal, diagnostics) => analyzeRows(selectedRows(selection), question, signal, diagnostics),
|
|
context(selection, question) {
|
|
const evidence = selectedRows(selection).flatMap(row => {
|
|
const analysis = cached(row, question);
|
|
return analysis ? [{id: row.id, messageId: row.messageId, position: row.position, scope: row.scope, label: row.label, analysis}] : [];
|
|
});
|
|
if (!evidence.length) return null;
|
|
const currentLabels = evidence.filter(image => image.scope === "current").map(image => image.label).join(", ");
|
|
return [
|
|
{role: "system", content: VISION_EVIDENCE_GUIDANCE + (currentLabels
|
|
? `\nThe following images are attached to the user's current message: ${currentLabels}.`
|
|
: "\nThe following evidence refers to earlier messages; the current user message has no image attachment.")},
|
|
{role: "user", content: "<image_evidence>\n" + JSON.stringify({question: normalizeVisionQuestion(question), images: evidence}).replaceAll("<", "\\u003c").replaceAll(">", "\\u003e") + "\n</image_evidence>"}
|
|
];
|
|
}
|
|
};
|
|
}
|