feat: complete v0.6 content and attachment experience

worker/index.ts

@@ -1,25 +1,60 @@
-import { readFile } from "node:fs/promises";
-import { join } from "node:path";
+import { readFile, mkdir, readdir, unlink, writeFile } from "node:fs/promises";
+import { extname, join } from "node:path";
+import { createHash } from "node:crypto";
 import { db } from "../lib/db";
 import { decrypt } from "../lib/crypto";
 import { createMemo, createRemoteFile, getMemosIdentity, listMemos, memoUrl, setMemoAttachments } from "../lib/memos";
 import { recordError } from "../lib/observability";
 
-type Source = { id: number; user_id: number; base_url: string; token_encrypted: string; is_enabled: number; sync_tags_json: string; sync_from: string | null; sync_to: string | null; sync_attachment_mode: "all" | "images" | "none" };
+type Source = { id: number; user_id: number; base_url: string; token_encrypted: string; is_enabled: number; sync_tags_json: string; sync_from: string | null; sync_to: string | null; sync_attachment_mode: "all" | "images" | "none"; attachment_storage_mode: "remote" | "images" | "all"; attachment_cache_limit_bytes: number };
 type Job = { id: number; source_id: number; kind: "pull" | "push"; payload_json: string | null; attempts: number };
 
-function upsertRemote(source: Source, memo: any) {
-  const tags = JSON.stringify(memo.tags || []), attachments = JSON.stringify(memo.attachments || memo.resources || []);
+function remoteAttachmentUrl(attachment: any, baseUrl: string) {
+  if (attachment.url || attachment.externalLink) return attachment.url || attachment.externalLink;
+  if (!attachment.name || !attachment.filename) return null;
+  return `${baseUrl.replace(/\/$/, "")}/file/${attachment.name.split("/").map(encodeURIComponent).join("/")}/${encodeURIComponent(attachment.filename)}`;
+}
+
+async function cacheAttachments(source: Source, attachments: any[]) {
+  if (source.attachment_storage_mode === "remote") return attachments;
+  const directory = join(process.cwd(), "public", "uploads", "cache", `source-${source.id}`); await mkdir(directory, { recursive: true });
+  let used = 0;
+  for (const row of db.prepare("SELECT attachments_json FROM posts WHERE source_id=?").all(source.id) as { attachments_json: string }[]) { try { used += (JSON.parse(row.attachments_json) as any[]).filter((item) => String(item.url || "").startsWith(`/uploads/cache/source-${source.id}/`)).reduce((sum, item) => sum + Number(item.size || 0), 0); } catch {} }
+  const result: any[] = [];
+  for (const attachment of attachments) {
+    const url = remoteAttachmentUrl(attachment, source.base_url); const type = attachment.type || "";
+    if (!url || (source.attachment_storage_mode === "images" && !type.startsWith("image/"))) { result.push(attachment); continue; }
+    try {
+      const target = new URL(url); if (target.origin !== new URL(source.base_url).origin) throw new Error("Attachment host is not the source host");
+      const response = await fetch(url, { signal: AbortSignal.timeout(15_000) }); if (!response.ok) throw new Error(`Attachment download ${response.status}`);
+      const body = Buffer.from(await response.arrayBuffer()); if (used + body.length > source.attachment_cache_limit_bytes) throw new Error("Attachment cache quota exceeded");
+      const filename = attachment.filename || attachment.name || "attachment"; const key = createHash("sha256").update(url).digest("hex").slice(0, 24) + extname(filename);
+      await writeFile(join(directory, key), body); used += body.length;
+      result.push({ ...attachment, originalUrl: url, url: `/uploads/cache/source-${source.id}/${key}`, type: type || response.headers.get("content-type") || "application/octet-stream", size: body.length });
+    } catch (error) { recordError("attachment-cache", error, { sourceId: source.id, url }); result.push(attachment); }
+  }
+  return result;
+}
+
+async function cleanupCache(source: Source) {
+  const directory = join(process.cwd(), "public", "uploads", "cache", `source-${source.id}`); let names: string[]; try { names = await readdir(directory); } catch { return; }
+  const used = new Set<string>(); for (const row of db.prepare("SELECT attachments_json FROM posts WHERE source_id=?").all(source.id) as { attachments_json: string }[]) { try { for (const attachment of JSON.parse(row.attachments_json) as any[]) { const url = String(attachment.url || ""); if (url.startsWith(`/uploads/cache/source-${source.id}/`)) used.add(url.split("/").at(-1)!); } } catch {} }
+  await Promise.all(names.filter((name) => !used.has(name)).map((name) => unlink(join(directory, name)).catch(() => undefined)));
+}
+
+async function upsertRemote(source: Source, memo: any) {
+  const tags = JSON.stringify(memo.tags || []), attachments = JSON.stringify(await cacheAttachments(source, memo.attachments || memo.resources || []));
   db.prepare(`INSERT INTO posts(source_id,author_id,remote_memo_name,content,visibility,tags_json,attachments_json,origin,remote_created_at,remote_updated_at,sync_status,hidden,remote_url) VALUES(?,?,?,?,?,?,?,?,?,?, 'synced',0,?) ON CONFLICT(source_id,remote_memo_name) DO UPDATE SET content=excluded.content,visibility=excluded.visibility,tags_json=excluded.tags_json,attachments_json=excluded.attachments_json,remote_updated_at=excluded.remote_updated_at,remote_url=excluded.remote_url,hidden=0,updated_at=CURRENT_TIMESTAMP`).run(source.id, source.user_id, memo.name, memo.content, memo.visibility, tags, attachments, "memos", memo.createTime || null, memo.updateTime || null, memoUrl(source.base_url, memo.name));
 }
 
 async function pull(source: Source) {
   const token = decrypt(source.token_encrypted); const rules = { tags: JSON.parse(source.sync_tags_json || "[]") as string[], from: source.sync_from, to: source.sync_to, attachmentMode: source.sync_attachment_mode };
   const [memos, identity] = await Promise.all([listMemos(source.base_url, token, rules), getMemosIdentity(source.base_url, token)]);
-  for (const memo of memos) upsertRemote(source, memo);
+  for (const memo of memos) await upsertRemote(source, memo);
   const names = memos.map((memo) => memo.name);
   if (names.length) { const placeholders = names.map(() => "?").join(","); db.prepare(`UPDATE posts SET hidden=1,updated_at=CURRENT_TIMESTAMP WHERE source_id=? AND remote_memo_name IS NOT NULL AND remote_memo_name NOT IN (${placeholders})`).run(source.id, ...names); }
   else db.prepare("UPDATE posts SET hidden=1,updated_at=CURRENT_TIMESTAMP WHERE source_id=? AND remote_memo_name IS NOT NULL").run(source.id);
+  await cleanupCache(source);
   const avatar = identity.avatarUrl || identity.avatar || null; const avatarUrl = avatar?.startsWith("/") ? `${source.base_url.replace(/\/$/, "")}${avatar}` : avatar;
   db.prepare("UPDATE sources SET sync_status='synced',last_synced_at=CURRENT_TIMESTAMP,last_error=NULL,last_connection_at=CURRENT_TIMESTAMP,last_connection_error=NULL,remote_display_name=?,remote_avatar_url=? WHERE id=?").run(identity.nickname || identity.username || identity.name, avatarUrl, source.id);
 }