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

feat: complete v0.2 source and sync management

Browse Source
This commit is contained in:
tangsongdayo
2026-07-19 01:33:45 +08:00
parent 4e9a13981d
commit 5b5129fad5
21 changed files with 367 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
worker/index.ts
+61 -8
View File
@@ -1,8 +1,61 @@
import { db } from "../lib/db"; import { decrypt } from "../lib/crypto"; import { createMemo, createRemoteFile, listMemos, setMemoAttachments } from "../lib/memos"; import { readFile } from "node:fs/promises"; import { join } from "node:path";
type Source={id:number;user_id:number;base_url:string;token_encrypted:string};
function upsertRemote(source:Source,m:any){const tags=JSON.stringify(m.tags||[]),attachments=JSON.stringify(m.attachments||m.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) 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,hidden=0,updated_at=CURRENT_TIMESTAMP`).run(source.id,source.user_id,m.name,m.content,m.visibility,tags,attachments,'memos',m.createTime||null,m.updateTime||null);}
async function pull(source:Source){const memos=await listMemos(source.base_url,decrypt(source.token_encrypted));for(const memo of memos)upsertRemote(source,memo);const names=memos.map(m=>m.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);}db.prepare("UPDATE sources SET sync_status='synced',last_synced_at=CURRENT_TIMESTAMP,last_error=NULL WHERE id=?").run(source.id);}
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=[] as unknown[],resources=[] as 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=?,sync_status='synced',updated_at=CURRENT_TIMESTAMP WHERE id=?").run(memo.name,memo.createTime||null,memo.updateTime||null,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 any;if(!job)return;db.prepare("UPDATE sync_jobs SET status='running',attempts=attempts+1 WHERE id=?").run(job.id);const source=db.prepare('SELECT * FROM sources WHERE id=?').get(job.source_id) as Source;try{if(job.kind==='pull')await pull(source);else if(job.kind==='push')await push(source,JSON.parse(job.payload_json||'{}'));db.prepare("UPDATE sync_jobs SET status='done' 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);}catch(error){const message=error instanceof Error?error.message:'Sync failure';db.prepare("UPDATE sync_jobs SET status=CASE WHEN attempts>=5 THEN 'failed' ELSE 'queued' END,last_error=?,run_after=datetime('now','+5 minutes') WHERE id=?").run(message,job.id);db.prepare("UPDATE sources SET sync_status='error',last_error=? WHERE id=?").run(message,source.id);}}
function schedule(){const interval=Number(process.env.SYNC_INTERVAL_MINUTES||60);db.prepare("INSERT INTO sync_jobs(source_id,kind) SELECT id,'pull' FROM sources WHERE 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`);}
setInterval(()=>{schedule();void run();},5000);schedule();void run();
import { readFile } from "node:fs/promises";
import { join } from "node:path";
import { db } from "../lib/db";
import { decrypt } from "../lib/crypto";
import { createMemo, createRemoteFile, listMemos, setMemoAttachments } from "../lib/memos";
type Source = { id: number; user_id: number; base_url: string; token_encrypted: string; is_enabled: 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 || []);
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) 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,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);
}
async function pull(source: Source) {
const memos = await listMemos(source.base_url, decrypt(source.token_encrypted));
for (const memo of memos) 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);
db.prepare("UPDATE sources SET sync_status='synced',last_synced_at=CURRENT_TIMESTAMP,last_error=NULL WHERE id=?").run(source.id);
}
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=?,sync_status='synced',updated_at=CURRENT_TIMESTAMP WHERE id=?").run(memo.name, memo.createTime || null, memo.updateTime || null, 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 {
if (job.kind === "pull") await pull(source); else 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);
} catch (error) {
const message = error instanceof Error ? error.message : "Sync failure"; const exhausted = job.attempts + 1 >= 5;
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=? WHERE id=?").run(message, source.id);
}
}
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`);
}
setInterval(() => { schedule(); void run(); }, 5000); schedule(); void run();