Mebbling/worker/index.ts

import { readFile, mkdir, readdir, unlink, writeFile } from "node:fs/promises";
import { extname, join } from "node:path";
import { createHash, randomUUID } 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";
import { fetchRss } from "../lib/rss";
import { sendAlert } from "../lib/alerts";

type Source = { id: number; user_id: number; base_url: string; token_encrypted: string; remote_user: string | null; integration_type: "memos" | "rss"; rss_feed_url: string | null; 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; sync_batch_size: number; sync_max_posts: number | null; sync_cursor: string | null; sync_imported_count: number; sync_run_id: string | null };
type Job = { id: number; source_id: number; kind: "pull" | "push"; payload_json: string | null; attempts: number };

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 { items: attachments, errors: [] as string[] };
  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[] = [], errors: string[] = [];
  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) { const message = error instanceof Error ? error.message : "快取附件失敗"; errors.push(message); recordError("attachment-cache", error, { sourceId: source.id, url }); result.push(attachment); }
  }
  return { items: result, errors };
}

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, syncRun: string) {
  const cached = await cacheAttachments(source, memo.attachments || memo.resources || []); const tags = JSON.stringify(memo.tags || []), attachments = JSON.stringify(cached.items);
  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,last_seen_sync_run) 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,last_seen_sync_run=excluded.last_seen_sync_run,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), syncRun);
  return cached.errors;
}

async function pull(source: Source) {
  if (source.integration_type === "rss") { const items = await fetchRss(source.rss_feed_url || ""); for (const item of items) 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(?,?,?,?,?,'[]','[]','rss',?,?, 'synced',0,?) ON CONFLICT(source_id,remote_memo_name) DO UPDATE SET content=excluded.content,remote_created_at=excluded.remote_created_at,remote_url=excluded.remote_url,hidden=0,updated_at=CURRENT_TIMESTAMP").run(source.id,source.user_id,`rss:${item.id}`,item.content,"PUBLIC",item.publishedAt,item.publishedAt,item.link); const names=items.map((item)=>`rss:${item.id}`); if(names.length) db.prepare(`UPDATE posts SET hidden=1 WHERE source_id=? AND remote_memo_name NOT IN (${names.map(()=>"?").join(",")})`).run(source.id,...names); db.prepare("UPDATE sources SET sync_status='synced',last_synced_at=CURRENT_TIMESTAMP,last_error=NULL WHERE id=?").run(source.id); return false; }
  const token = decrypt(source.token_encrypted); const identity = await getMemosIdentity(source.base_url, token); const creator = identity.name;
  const rules = { creator, tags: JSON.parse(source.sync_tags_json || "[]") as string[], from: source.sync_from, to: source.sync_to, attachmentMode: source.sync_attachment_mode };
  const remaining = source.sync_max_posts ? Math.max(0, source.sync_max_posts - source.sync_imported_count) : source.sync_batch_size;
  if (remaining === 0) { db.prepare("UPDATE sources SET sync_cursor=NULL,sync_status='synced',last_synced_at=CURRENT_TIMESTAMP WHERE id=?").run(source.id); return false; }
  const page = await listMemos(source.base_url, token, rules, { pageToken: source.sync_cursor || "", pageSize: Math.min(source.sync_batch_size, remaining) }); const memos = page.memos; const syncRun = source.sync_run_id || randomUUID();
  const cacheErrors: string[] = []; for (const memo of memos) cacheErrors.push(...await upsertRemote(source, memo, syncRun));
  const imported = source.sync_imported_count + memos.length; const capped = Boolean(source.sync_max_posts && imported >= source.sync_max_posts); const complete = !page.nextPageToken && !capped; const more = Boolean(page.nextPageToken) && !capped;
  if (complete) db.prepare("UPDATE posts SET hidden=1,updated_at=CURRENT_TIMESTAMP WHERE source_id=? AND origin='memos' AND COALESCE(last_seen_sync_run,'')<>?").run(source.id, syncRun);
  await cleanupCache(source);
  const avatar = identity.avatarUrl || identity.avatar || null; const avatarUrl = avatar?.startsWith("/") ? `${source.base_url.replace(/\/$/, "")}${avatar}` : avatar;
  const name = identity.displayName || identity.nickname || identity.username || identity.name;
  db.prepare("UPDATE sources SET name=?,remote_user=?,sync_status=?,last_synced_at=CASE WHEN ? THEN NULL ELSE CURRENT_TIMESTAMP END,last_error=NULL,last_connection_at=CURRENT_TIMESTAMP,last_connection_error=NULL,attachment_cache_error=?,remote_display_name=?,remote_avatar_url=?,sync_cursor=?,sync_imported_count=?,sync_run_id=? WHERE id=?").run(name, creator, more ? 'importing' : 'synced', more ? 1 : 0, cacheErrors.length ? `${cacheErrors.length} 個附件快取失敗；可按「立即同步」重試。` : null, name, avatarUrl, more ? page.nextPageToken : null, more ? imported : 0, more ? syncRun : null, source.id);
  return more;
}

async function push(source: Source, payload: any) {
  const post = db.prepare("SELECT * FROM posts WHERE id=? AND source_id=?").get(payload.postId, source.id) as any;
  if (!post) return;
  const token = decrypt(source.token_encrypted); const localAttachments = JSON.parse(post.attachments_json || "[]") as { name: string; url: string; type: string; size: number }[];
  const attachments: unknown[] = [], resources: unknown[] = [];
  for (const attachment of localAttachments) {
    const content = (await readFile(join(process.cwd(), "public", attachment.url))).toString("base64");
    const remote = await createRemoteFile(source.base_url, token, { filename: attachment.name, content, type: attachment.type || "application/octet-stream", size: String(attachment.size) });
    (remote.kind === "attachment" ? attachments : resources).push(remote.value);
  }
  const memo = await createMemo(source.base_url, token, { content: post.content, visibility: post.visibility, resources });
  if (attachments.length) await setMemoAttachments(source.base_url, token, memo.name, attachments);
  db.prepare("UPDATE posts SET remote_memo_name=?,remote_created_at=?,remote_updated_at=?,remote_url=?,sync_status='synced',updated_at=CURRENT_TIMESTAMP WHERE id=?").run(memo.name, memo.createTime || null, memo.updateTime || null, memoUrl(source.base_url, memo.name), post.id);
}

async function run() {
  const job = db.prepare("SELECT * FROM sync_jobs WHERE status='queued' AND run_after<=CURRENT_TIMESTAMP ORDER BY id LIMIT 1").get() as Job | undefined;
  if (!job) return;
  db.prepare("UPDATE sync_jobs SET status='running',attempts=attempts+1,started_at=CURRENT_TIMESTAMP WHERE id=?").run(job.id);
  const source = db.prepare("SELECT * FROM sources WHERE id=?").get(job.source_id) as Source | undefined;
  if (!source || !source.is_enabled) { db.prepare("UPDATE sync_jobs SET status='cancelled',last_error='Source is disabled or deleted',finished_at=CURRENT_TIMESTAMP WHERE id=?").run(job.id); return; }
  try {
    const more = job.kind === "pull" ? await pull(source) : false; if (job.kind === "push") await push(source, JSON.parse(job.payload_json || "{}"));
    db.prepare("UPDATE sync_jobs SET status='done',finished_at=CURRENT_TIMESTAMP WHERE id=?").run(job.id);
    if (more) db.prepare("INSERT INTO sync_jobs(source_id,kind,trigger) VALUES(?,'pull','batch')").run(source.id); else db.prepare("UPDATE sources SET sync_status='synced',last_error=NULL,last_synced_at=CURRENT_TIMESTAMP WHERE id=?").run(source.id);
  } catch (error) {
    const message = error instanceof Error ? error.message : "Sync failure"; const exhausted = job.attempts + 1 >= 5;
    recordError("sync", error, { sourceId: source.id, jobId: job.id, kind: job.kind, attempts: job.attempts });
    db.prepare("UPDATE sync_jobs SET status=?,last_error=?,finished_at=CASE WHEN ? THEN CURRENT_TIMESTAMP ELSE NULL END,run_after=CASE WHEN ? THEN run_after ELSE datetime('now','+5 minutes') END WHERE id=?").run(exhausted ? "failed" : "queued", message, exhausted ? 1 : 0, exhausted ? 1 : 0, job.id);
    db.prepare("UPDATE sources SET sync_status='error',last_error=?,last_connection_error=? WHERE id=?").run(message, message, source.id);
    if (exhausted) void sendAlert(`sync:${source.id}`, "Mebbling 同步失敗", `來源 #${source.id} 已重試 5 次仍失敗：${message}`);
  }
}

function schedule() {
  const interval = Number(process.env.SYNC_INTERVAL_MINUTES || 60);
  db.prepare(`INSERT INTO sync_jobs(source_id,kind,trigger) SELECT id,'pull','scheduled' FROM sources WHERE is_enabled=1 AND COALESCE(last_synced_at,'1970-01-01') < datetime('now', ?) AND NOT EXISTS (SELECT 1 FROM sync_jobs j WHERE j.source_id=sources.id AND j.kind='pull' AND j.status IN ('queued','running'))`).run(`-${interval} minutes`);
}

function retention() {
  const days = (name: string) => Math.max(0, Number(process.env[name] || 0) || 0);
  db.prepare("DELETE FROM source_invites WHERE expires_at<CURRENT_TIMESTAMP AND (used_at IS NOT NULL OR expires_at<datetime('now','-30 days'))").run();
  const cleanup = (table: string, column: string, value: number) => { if (value) db.prepare(`DELETE FROM ${table} WHERE ${column}<datetime('now', ?)`).run(`-${value} days`); };
  cleanup("notifications", "created_at", days("NOTIFICATION_RETENTION_DAYS")); cleanup("reading_history", "last_read_at", days("READING_HISTORY_RETENTION_DAYS")); cleanup("audit_events", "created_at", days("AUDIT_RETENTION_DAYS"));
}

function webhookHealth() {
  const stale = db.prepare("SELECT id,name,last_webhook_at FROM sources WHERE is_enabled=1 AND webhook_mode='signed' AND (last_webhook_at IS NULL OR last_webhook_at<datetime('now','-7 days'))").all() as { id: number; name: string; last_webhook_at: string | null }[];
  for (const source of stale) void sendAlert(`webhook:${source.id}`, "Mebbling Webhook 未收到事件", `來源「${source.name}」超過 7 天未收到 Webhook；目前仍會以定期 API 同步校正。`);
}

let lastRetention = 0, lastWebhookHealth = 0; setInterval(() => { schedule(); if (Date.now() - lastRetention > 86_400_000) { retention(); lastRetention = Date.now(); } if (Date.now() - lastWebhookHealth > 3_600_000) { webhookHealth(); lastWebhookHealth = Date.now(); } void run(); }, 5000); schedule(); retention(); webhookHealth(); void run();