tangsongdayo/Mebbling
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

feat: batch large source imports

Browse Source
This commit is contained in:
tangsongdayo
2026-07-19 05:43:03 +08:00
parent 591981b7c8
commit 9e159910ca
8 changed files with 40 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files
worker/index.ts
+15 -13
View File
@@ -1,13 +1,13 @@
import { readFile, mkdir, readdir, unlink, writeFile } from "node:fs/promises";
import { extname, join } from "node:path";
import { createHash } from "node:crypto";
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";
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 };
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) {
@@ -43,25 +43,27 @@ async function cleanupCache(source: Source) {
await Promise.all(names.filter((name) => !used.has(name)).map((name) => unlink(join(directory, name)).catch(() => undefined)));
}
async function upsertRemote(source: Source, memo: any) {
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) 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));
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; }
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 memos = await listMemos(source.base_url, token, rules);
const cacheErrors: string[] = []; for (const memo of memos) cacheErrors.push(...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);
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='synced',last_synced_at=CURRENT_TIMESTAMP,last_error=NULL,last_connection_at=CURRENT_TIMESTAMP,last_connection_error=NULL,attachment_cache_error=?,remote_display_name=?,remote_avatar_url=? WHERE id=?").run(name, creator, cacheErrors.length ? `${cacheErrors.length} 個附件快取失敗；可按「立即同步」重試。` : null, name, avatarUrl, source.id);
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) {
@@ -86,9 +88,9 @@ async function run() {
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 {
if (job.kind === "pull") await pull(source); else if (job.kind === "push") await push(source, JSON.parse(job.payload_json || "{}"));
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);
db.prepare("UPDATE sources SET sync_status='synced',last_error=NULL,last_synced_at=CURRENT_TIMESTAMP WHERE id=?").run(source.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 });