diff --git a/.gitignore b/.gitignore
index 4f1d2a1..4c232e3 100644
--- a/.gitignore
+++ b/.gitignore
@@ -4,5 +4,6 @@ node_modules/
.env
data/*.db
data/*.db-*
+data/backups/
public/uploads/*
!public/uploads/.gitkeep
diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
index 0000000..f1db8bb
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,24 @@
+# Changelog
+
+本專案遵循 [Semantic Versioning](https://semver.org/lang/zh-TW/);版本 `0.x` 表示功能仍可能調整。
+
+## [0.2.0] - Unreleased
+
+### Added
+
+- 來源的重新命名、停用/啟用、刪除、離開共享來源與建立者轉移。
+- 同步工作觸發來源、開始/完成時間、重試次數與控制台歷史紀錄。
+- Pull 同步去重,避免手動、webhook 與排程重複建立處理中工作。
+- 帳號密碼變更與管理員同步異常檢視頁。
+- SQLite migration 版本紀錄,以及資料庫與附件備份/還原文件和腳本。
+
+### Changed
+
+- 停用來源後不再接受 webhook、手動同步、排程同步或 Hub 發文推送。
+- 來源刪除時會移除遠端鏡像貼文,保留 Hub 原生貼文但解除來源關聯。
+
+## [0.1.0] - 2026-07-19
+
+### Added
+
+- 第一個公開 Pre-release:Memos 公開貼文聚合、附件、留言、表情、Hub 發文與 webhook 同步。
diff --git a/README.md b/README.md
index d38cdef..b78c898 100644
--- a/README.md
+++ b/README.md
@@ -2,6 +2,8 @@
自架的 Memos 公開貼文 Hub。將朋友各自 Memos 中的公開貼文集中展示,同時保留 Hub 內的留言、表情回應與發文功能。
+目前開發版本:`v0.2.0`(尚未發布)。版本變更請見 [CHANGELOG.md](CHANGELOG.md)。
+
## 功能
- 匯入多個 Memos 來源的 `PUBLIC` 貼文、標籤與附件。
@@ -11,6 +13,9 @@
- 使用者可加入共享來源並手動同步或發文;僅來源建立者能管理 webhook URL。
- Webhook URL 採不可猜測的隨機密鑰路徑、雜湊保存與簡易速率限制。
- 控制台會顯示最近一次收到 webhook 的時間及最後同步時間。
+- 來源建立者可重新命名、停用、刪除或轉移所有權;共享成員可自行離開來源。
+- 同步工作具去重、重試、觸發來源與歷史紀錄;管理員可集中檢視異常。
+- 內建 SQLite 與附件備份腳本,以及可追蹤的 schema migration。
## 快速啟動(WSL/Docker)
@@ -113,6 +118,8 @@ https://你的網域/api/sync/webhook/來源ID/隨機密鑰
`data/` 與 `public/uploads/` 是正式資料,備份時請一併備份。`.next/` 與 `node_modules/` 是可重新產生的建置/依賴資料,不需備份。
+備份、還原及資料庫 migration 的操作請見 [維運文件](docs/OPERATIONS.md)。
+
## 正式部署
將 `NEXT_PUBLIC_APP_URL` 設成實際 HTTPS 網域,並以反向代理將該網域導向 Web 容器的 3000 連接埠(或主機的 8088 對應埠)。務必確保外部可連到 webhook URL,否則仍會由定期校正同步補回資料,但不會即時更新。
diff --git a/app/account/page.tsx b/app/account/page.tsx
new file mode 100644
index 0000000..3ed5e89
--- /dev/null
+++ b/app/account/page.tsx
@@ -0,0 +1,7 @@
+import { redirect } from "next/navigation";
+import { getSession } from "@/lib/auth";
+
+export default async function Account({ searchParams }: { searchParams: Promise<{ error?: string; updated?: string }> }) {
+ const user = await getSession(); if (!user) redirect("/login"); const query = await searchParams;
+ return <>
帳號設定 {query.error && {query.error}
}{query.updated && 密碼已更新。
}>;
+}
diff --git a/app/admin/page.tsx b/app/admin/page.tsx
new file mode 100644
index 0000000..51b0557
--- /dev/null
+++ b/app/admin/page.tsx
@@ -0,0 +1,12 @@
+import { redirect } from "next/navigation";
+import { getSession } from "@/lib/auth";
+import { db } from "@/lib/db";
+
+export const dynamic = "force-dynamic";
+
+export default async function AdminPage() {
+ const user = await getSession(); if (!user || user.role !== "admin") redirect("/");
+ const failures = db.prepare("SELECT j.id,j.kind,j.trigger,j.status,j.attempts,j.last_error,j.created_at,j.finished_at,s.id AS source_id,s.name,s.base_url FROM sync_jobs j JOIN sources s ON s.id=j.source_id WHERE j.status='failed' OR s.sync_status='error' ORDER BY COALESCE(j.finished_at,j.created_at) DESC LIMIT 100").all() as any[];
+ const sources = db.prepare("SELECT s.id,s.name,s.base_url,s.sync_status,s.last_synced_at,s.last_error,s.is_enabled,count(sm.user_id) AS member_count FROM sources s LEFT JOIN source_members sm ON sm.source_id=s.id GROUP BY s.id ORDER BY s.id DESC").all() as any[];
+ return <>管理員:同步狀態 失敗或異常工作 {failures.length ? {failures.map((item) => {item.name} (來源 #{item.source_id}) · {item.kind || "source"} · {item.status || "error"} {item.last_error || "來源處於錯誤狀態"} 嘗試 {item.attempts || 0} 次;{new Date((item.finished_at || item.created_at) + "Z").toLocaleString("zh-TW")} )} : 沒有同步異常。
}所有來源 {sources.map((source) => {source.name} · {source.is_enabled ? source.sync_status : "disabled"} · 成員 {source.member_count}#{source.id} · {source.base_url} · 上次同步:{source.last_synced_at || "尚未完成"} {source.last_error && <>{source.last_error} >} )} >;
+}
diff --git a/app/api/auth/password/route.ts b/app/api/auth/password/route.ts
new file mode 100644
index 0000000..a34f7f2
--- /dev/null
+++ b/app/api/auth/password/route.ts
@@ -0,0 +1,18 @@
+import bcrypt from "bcryptjs";
+import { NextResponse } from "next/server";
+import { requireUser } from "@/lib/auth";
+import { db } from "@/lib/db";
+import { externalUrl } from "@/lib/http";
+
+export async function POST(req: Request) {
+ try {
+ const user = await requireUser(); const form = await req.formData();
+ const currentPassword = String(form.get("currentPassword") || ""); const newPassword = String(form.get("newPassword") || ""); const confirmPassword = String(form.get("confirmPassword") || "");
+ if (newPassword.length < 10) throw new Error("新密碼至少需要 10 個字元");
+ if (newPassword !== confirmPassword) throw new Error("兩次新密碼不一致");
+ const account = db.prepare("SELECT password_hash FROM users WHERE id=? AND disabled=0").get(user.id) as { password_hash: string } | undefined;
+ if (!account || !(await bcrypt.compare(currentPassword, account.password_hash))) throw new Error("目前密碼不正確");
+ db.prepare("UPDATE users SET password_hash=? WHERE id=?").run(await bcrypt.hash(newPassword, 12), user.id);
+ return NextResponse.redirect(externalUrl(req, "/account?updated=1"));
+ } catch (error) { return NextResponse.redirect(externalUrl(req, "/account?error=" + encodeURIComponent(error instanceof Error ? error.message : "password"))); }
+}
diff --git a/app/api/posts/route.ts b/app/api/posts/route.ts
index a71f2fd..57cbe56 100644
--- a/app/api/posts/route.ts
+++ b/app/api/posts/route.ts
@@ -1,2 +1,2 @@
import { NextResponse } from "next/server"; import { requireUser } from "@/lib/auth"; import { db } from "@/lib/db"; import { externalUrl } from "@/lib/http"; import { mkdir, writeFile } from "node:fs/promises"; import { randomUUID } from "node:crypto"; import { extname, join } from "node:path";
-export async function POST(req:Request){const json=req.headers.get("accept")?.includes("application/json");try{const user=await requireUser();const f=await req.formData();const content=String(f.get("content")||"").trim();const visibility=String(f.get("visibility")||"PUBLIC");const sourceId=Number(f.get("sourceId"));const tags=String(f.get("tags")||"").split(/\s*,\s*/).filter(Boolean).map(t=>t.replace(/^#/,""));if(!content||!['PRIVATE','PROTECTED','PUBLIC'].includes(visibility)||!sourceId)throw new Error("Invalid post");const source=db.prepare("SELECT s.id FROM sources s JOIN source_members sm ON sm.source_id=s.id WHERE s.id=? AND sm.user_id=?").get(sourceId,user.id);if(!source)throw new Error("Source not found");const max=Number(process.env.UPLOAD_MAX_BYTES||10485760);const files=f.getAll('attachments').filter((x):x is File=>x instanceof File&&x.size>0);const attachments:any[]=[];await mkdir(join(process.cwd(),'public','uploads'),{recursive:true});for(const file of files){if(file.size>max)throw new Error(`${file.name} exceeds upload limit`);const id=randomUUID()+extname(file.name);await writeFile(join(process.cwd(),'public','uploads',id),Buffer.from(await file.arrayBuffer()));attachments.push({name:file.name,url:`/uploads/${id}`,type:file.type,size:file.size});}const out=db.prepare("INSERT INTO posts(source_id,author_id,content,visibility,tags_json,attachments_json,origin,sync_status) VALUES(?,?,?,?,?,?,'hub','queued')").run(sourceId,user.id,content,visibility,JSON.stringify(tags),JSON.stringify(attachments));db.prepare("INSERT INTO sync_jobs(source_id,kind,payload_json) VALUES(?, 'push', ?)").run(sourceId,JSON.stringify({postId:out.lastInsertRowid}));if(json)return NextResponse.json({id:Number(out.lastInsertRowid)},{status:201});return NextResponse.redirect(externalUrl(req,`/posts/${out.lastInsertRowid}`));}catch(e){const message=e instanceof Error?e.message:'post';if(json)return NextResponse.json({error:message},{status:400});return NextResponse.redirect(externalUrl(req,'/dashboard?error='+encodeURIComponent(message)));}}
+export async function POST(req:Request){const json=req.headers.get("accept")?.includes("application/json");try{const user=await requireUser();const f=await req.formData();const content=String(f.get("content")||"").trim();const visibility=String(f.get("visibility")||"PUBLIC");const sourceId=Number(f.get("sourceId"));const tags=String(f.get("tags")||"").split(/\s*,\s*/).filter(Boolean).map(t=>t.replace(/^#/,""));if(!content||!['PRIVATE','PROTECTED','PUBLIC'].includes(visibility)||!sourceId)throw new Error("Invalid post");const source=db.prepare("SELECT s.id FROM sources s JOIN source_members sm ON sm.source_id=s.id WHERE s.id=? AND sm.user_id=? AND s.is_enabled=1").get(sourceId,user.id);if(!source)throw new Error("Source not available");const max=Number(process.env.UPLOAD_MAX_BYTES||10485760);const files=f.getAll('attachments').filter((x):x is File=>x instanceof File&&x.size>0);const attachments:any[]=[];await mkdir(join(process.cwd(),'public','uploads'),{recursive:true});for(const file of files){if(file.size>max)throw new Error(`${file.name} exceeds upload limit`);const id=randomUUID()+extname(file.name);await writeFile(join(process.cwd(),'public','uploads',id),Buffer.from(await file.arrayBuffer()));attachments.push({name:file.name,url:`/uploads/${id}`,type:file.type,size:file.size});}const out=db.prepare("INSERT INTO posts(source_id,author_id,content,visibility,tags_json,attachments_json,origin,sync_status) VALUES(?,?,?,?,?,?,'hub','queued')").run(sourceId,user.id,content,visibility,JSON.stringify(tags),JSON.stringify(attachments));db.prepare("INSERT INTO sync_jobs(source_id,kind,payload_json,trigger) VALUES(?, 'push', ?, 'manual')").run(sourceId,JSON.stringify({postId:out.lastInsertRowid}));if(json)return NextResponse.json({id:Number(out.lastInsertRowid)},{status:201});return NextResponse.redirect(externalUrl(req,`/posts/${out.lastInsertRowid}`));}catch(e){const message=e instanceof Error?e.message:'post';if(json)return NextResponse.json({error:message},{status:400});return NextResponse.redirect(externalUrl(req,'/dashboard?error='+encodeURIComponent(message)));}}
diff --git a/app/api/sources/[id]/manage/route.ts b/app/api/sources/[id]/manage/route.ts
new file mode 100644
index 0000000..fb20bac
--- /dev/null
+++ b/app/api/sources/[id]/manage/route.ts
@@ -0,0 +1,37 @@
+import { NextResponse } from "next/server";
+import { requireUser } from "@/lib/auth";
+import { db } from "@/lib/db";
+import { externalUrl } from "@/lib/http";
+import { queuePull } from "@/lib/sync";
+
+export async function POST(req: Request, { params }: { params: Promise<{ id: string }> }) {
+ try {
+ const user = await requireUser(); const { id: rawId } = await params; const id = Number(rawId); const form = await req.formData(); const action = String(form.get("action") || "");
+ const source = db.prepare("SELECT id,user_id FROM sources WHERE id=?").get(id) as { id: number; user_id: number } | undefined;
+ const member = db.prepare("SELECT role FROM source_members WHERE source_id=? AND user_id=?").get(id, user.id);
+ if (!source || !member) throw new Error("Source not found");
+ const owner = source.user_id === user.id;
+ if (action === "rename") {
+ const name = String(form.get("name") || "").trim(); if (!owner || !name || name.length > 80) throw new Error("Only the owner can rename a source");
+ db.prepare("UPDATE sources SET name=? WHERE id=?").run(name, id);
+ } else if (action === "set-enabled") {
+ if (!owner) throw new Error("Only the owner can change source status"); const enabled = String(form.get("enabled")) === "1";
+ db.prepare("UPDATE sources SET is_enabled=?,disabled_at=CASE WHEN ? THEN NULL ELSE CURRENT_TIMESTAMP END,sync_status=CASE WHEN ? THEN 'pending' ELSE 'disabled' END WHERE id=?").run(enabled ? 1 : 0, enabled ? 1 : 0, enabled ? 1 : 0, id);
+ if (enabled) queuePull(id, "manual");
+ } else if (action === "leave") {
+ if (owner) throw new Error("Transfer ownership or delete the source before leaving");
+ db.prepare("DELETE FROM source_members WHERE source_id=? AND user_id=?").run(id, user.id);
+ } else if (action === "transfer") {
+ if (!owner) throw new Error("Only the owner can transfer ownership"); const username = String(form.get("username") || "").trim();
+ const target = db.prepare("SELECT u.id FROM users u JOIN source_members sm ON sm.user_id=u.id WHERE sm.source_id=? AND u.username=?").get(id, username) as { id: number } | undefined;
+ if (!target || target.id === user.id) throw new Error("Choose another existing member");
+ const transfer = db.transaction(() => { db.prepare("UPDATE sources SET user_id=? WHERE id=?").run(target.id, id); db.prepare("UPDATE source_members SET role='member' WHERE source_id=? AND user_id=?").run(id, user.id); db.prepare("UPDATE source_members SET role='owner' WHERE source_id=? AND user_id=?").run(id, target.id); });
+ transfer();
+ } else if (action === "delete") {
+ if (!owner) throw new Error("Only the owner can delete a source");
+ const remove = db.transaction(() => { db.prepare("DELETE FROM posts WHERE source_id=? AND origin='memos'").run(id); db.prepare("UPDATE posts SET source_id=NULL WHERE source_id=? AND origin='hub'").run(id); db.prepare("DELETE FROM sources WHERE id=?").run(id); });
+ remove();
+ } else throw new Error("Unknown source action");
+ return NextResponse.redirect(externalUrl(req, "/dashboard?source=updated"));
+ } catch (error) { return NextResponse.redirect(externalUrl(req, "/dashboard?error=" + encodeURIComponent(error instanceof Error ? error.message : "source"))); }
+}
diff --git a/app/api/sources/route.ts b/app/api/sources/route.ts
index af5f473..89e4fef 100644
--- a/app/api/sources/route.ts
+++ b/app/api/sources/route.ts
@@ -1,8 +1,29 @@
-import { NextResponse } from "next/server"; import { requireUser } from "@/lib/auth"; import { decrypt, encrypt } from "@/lib/crypto"; import { getMemosIdentity, verifyMemos } from "@/lib/memos"; import { db } from "@/lib/db"; import { externalUrl } from "@/lib/http";
-export async function POST(req: Request) { try { const user=await requireUser(); const form=await req.formData(); const name=String(form.get("name")||"").trim(); const rawBaseUrl=String(form.get("baseUrl")||"").trim(); const token=String(form.get("token")||"").trim(); let baseUrl=""; try { const url=new URL(rawBaseUrl); if(!['http:','https:'].includes(url.protocol)) throw new Error(); baseUrl=`${url.origin}${url.pathname.replace(/\/+$/,"")}`; } catch { throw new Error("Invalid source URL"); } if(!name||token.length<20) throw new Error("Invalid source"); await verifyMemos(baseUrl,token); const identity=await getMemosIdentity(baseUrl,token);
- const legacySources=db.prepare("SELECT id,token_encrypted FROM sources WHERE base_url=? AND remote_user IS NULL").all(baseUrl) as {id:number;token_encrypted:string}[];
- for(const legacy of legacySources){try{const legacyIdentity=await getMemosIdentity(baseUrl,decrypt(legacy.token_encrypted));db.prepare("UPDATE sources SET remote_user=? WHERE id=? AND remote_user IS NULL").run(legacyIdentity.name,legacy.id);}catch{ /* Keep unavailable legacy sources unchanged. */ }}
- const shared=db.prepare("SELECT id FROM sources WHERE base_url=? AND remote_user=?").get(baseUrl,identity.name) as {id:number}|undefined;
- if(shared){db.prepare("INSERT OR IGNORE INTO source_members(source_id,user_id) VALUES(?,?)").run(shared.id,user.id);return NextResponse.redirect(externalUrl(req,"/dashboard?source=shared"));}
- const out=db.prepare("INSERT INTO sources(user_id,name,base_url,token_encrypted,remote_user,sync_status) VALUES(?,?,?,?,?, 'queued')").run(user.id,name,baseUrl,encrypt(token),identity.name);const sourceId=Number(out.lastInsertRowid);db.prepare("INSERT INTO source_members(source_id,user_id,role) VALUES(?,?,'owner')").run(sourceId,user.id);db.prepare("INSERT INTO sync_jobs(source_id,kind) VALUES(?, 'pull')").run(sourceId);return NextResponse.redirect(externalUrl(req,"/dashboard?source=connected"));
- } catch(e){ return NextResponse.redirect(externalUrl(req,"/dashboard?error="+encodeURIComponent(e instanceof Error?e.message:"source"))); } }
+import { NextResponse } from "next/server";
+import { requireUser } from "@/lib/auth";
+import { decrypt, encrypt } from "@/lib/crypto";
+import { db } from "@/lib/db";
+import { externalUrl } from "@/lib/http";
+import { getMemosIdentity, verifyMemos } from "@/lib/memos";
+import { queuePull } from "@/lib/sync";
+
+export async function POST(req: Request) {
+ try {
+ const user = await requireUser(); const form = await req.formData();
+ const name = String(form.get("name") || "").trim(); const rawBaseUrl = String(form.get("baseUrl") || "").trim(); const token = String(form.get("token") || "").trim();
+ let baseUrl = "";
+ try { const url = new URL(rawBaseUrl); if (!['http:', 'https:'].includes(url.protocol)) throw new Error(); baseUrl = `${url.origin}${url.pathname.replace(/\/+$/, "")}`; } catch { throw new Error("Invalid source URL"); }
+ if (!name || token.length < 20) throw new Error("Invalid source");
+ await verifyMemos(baseUrl, token); const identity = await getMemosIdentity(baseUrl, token);
+ const legacySources = db.prepare("SELECT id,token_encrypted FROM sources WHERE base_url=? AND remote_user IS NULL").all(baseUrl) as { id: number; token_encrypted: string }[];
+ for (const legacy of legacySources) {
+ try { const legacyIdentity = await getMemosIdentity(baseUrl, decrypt(legacy.token_encrypted)); db.prepare("UPDATE sources SET remote_user=? WHERE id=? AND remote_user IS NULL").run(legacyIdentity.name, legacy.id); } catch { /* Retry on a future connection. */ }
+ }
+ const shared = db.prepare("SELECT id FROM sources WHERE base_url=? AND remote_user=?").get(baseUrl, identity.name) as { id: number } | undefined;
+ if (shared) { db.prepare("INSERT OR IGNORE INTO source_members(source_id,user_id) VALUES(?,?)").run(shared.id, user.id); return NextResponse.redirect(externalUrl(req, "/dashboard?source=shared")); }
+ const out = db.prepare("INSERT INTO sources(user_id,name,base_url,token_encrypted,remote_user,sync_status) VALUES(?,?,?,?,?, 'queued')").run(user.id, name, baseUrl, encrypt(token), identity.name);
+ const sourceId = Number(out.lastInsertRowid);
+ db.prepare("INSERT INTO source_members(source_id,user_id,role) VALUES(?,?,'owner')").run(sourceId, user.id);
+ queuePull(sourceId, "source-created");
+ return NextResponse.redirect(externalUrl(req, "/dashboard?source=connected"));
+ } catch (error) { return NextResponse.redirect(externalUrl(req, "/dashboard?error=" + encodeURIComponent(error instanceof Error ? error.message : "source"))); }
+}
diff --git a/app/api/sync/route.ts b/app/api/sync/route.ts
index 61fe2a7..5bbe61a 100644
--- a/app/api/sync/route.ts
+++ b/app/api/sync/route.ts
@@ -1,2 +1,15 @@
-import { NextResponse } from "next/server"; import { requireUser } from "@/lib/auth"; import { db } from "@/lib/db"; import { externalUrl } from "@/lib/http";
-export async function POST(req:Request){try{const user=await requireUser();const f=await req.formData();const sourceId=Number(f.get('sourceId'));const source=db.prepare('SELECT s.id FROM sources s JOIN source_members sm ON sm.source_id=s.id WHERE s.id=? AND sm.user_id=?').get(sourceId,user.id);if(!source)throw 0;db.prepare("INSERT INTO sync_jobs(source_id,kind) VALUES(?, 'pull')").run(sourceId);return NextResponse.redirect(externalUrl(req,'/dashboard'));}catch{return NextResponse.redirect(externalUrl(req,'/'));}}
+import { NextResponse } from "next/server";
+import { requireUser } from "@/lib/auth";
+import { db } from "@/lib/db";
+import { externalUrl } from "@/lib/http";
+import { queuePull } from "@/lib/sync";
+
+export async function POST(req: Request) {
+ try {
+ const user = await requireUser(); const form = await req.formData(); const sourceId = Number(form.get("sourceId"));
+ const source = db.prepare("SELECT s.id FROM sources s JOIN source_members sm ON sm.source_id=s.id WHERE s.id=? AND sm.user_id=? AND s.is_enabled=1").get(sourceId, user.id);
+ if (!source) throw new Error("Source not available");
+ const created = queuePull(sourceId, "manual");
+ return NextResponse.redirect(externalUrl(req, `/dashboard?sync=${created ? "queued" : "already-queued"}`));
+ } catch (error) { return NextResponse.redirect(externalUrl(req, "/dashboard?error=" + encodeURIComponent(error instanceof Error ? error.message : "sync"))); }
+}
diff --git a/app/api/sync/webhook/[sourceId]/[secret]/route.ts b/app/api/sync/webhook/[sourceId]/[secret]/route.ts
index 17dfe89..b816eef 100644
--- a/app/api/sync/webhook/[sourceId]/[secret]/route.ts
+++ b/app/api/sync/webhook/[sourceId]/[secret]/route.ts
@@ -2,17 +2,18 @@ import { NextResponse } from "next/server";
import { db } from "@/lib/db";
import { withinRateLimit } from "@/lib/rate-limit";
import { webhookSecretMatches } from "@/lib/webhook";
+import { queuePull } from "@/lib/sync";
export async function POST(request: Request, { params }: { params: Promise<{ sourceId: string; secret: string }> }) {
const { sourceId, secret } = await params;
const id = Number(sourceId);
- const source = db.prepare("SELECT id, webhook_secret_hash FROM sources WHERE id=?").get(id) as { id: number; webhook_secret_hash: string | null } | undefined;
+ const source = db.prepare("SELECT id, webhook_secret_hash FROM sources WHERE id=? AND is_enabled=1").get(id) as { id: number; webhook_secret_hash: string | null } | undefined;
if (!source || !webhookSecretMatches(secret, source.webhook_secret_hash)) return NextResponse.json({ error: "Not found" }, { status: 404 });
const forwarded = request.headers.get("x-forwarded-for")?.split(",")[0].trim() || "unknown";
if (!withinRateLimit(`webhook:${id}:${forwarded}`)) return NextResponse.json({ error: "Too many requests" }, { status: 429 });
let payload: unknown = {};
try { payload = await request.json(); } catch { /* Memos payload is optional; a pull reconciles source state. */ }
db.prepare("UPDATE sources SET last_webhook_at=CURRENT_TIMESTAMP WHERE id=?").run(id);
- db.prepare("INSERT INTO sync_jobs(source_id,kind,payload_json) VALUES(?, 'pull', ?)").run(id, JSON.stringify(payload));
- return NextResponse.json({ ok: true });
+ const queued = queuePull(id, "webhook", payload);
+ return NextResponse.json({ ok: true, queued });
}
diff --git a/app/dashboard/page.tsx b/app/dashboard/page.tsx
index 2a556d0..18caf91 100644
--- a/app/dashboard/page.tsx
+++ b/app/dashboard/page.tsx
@@ -1,3 +1,35 @@
-import { redirect } from "next/navigation"; import { getSession } from "@/lib/auth"; import { db } from "@/lib/db"; import { PublishForm } from "./publish-form"; import { WebhookControl } from "./webhook-control";
-export const dynamic="force-dynamic";
-export default async function Dashboard({searchParams}:{searchParams:Promise<{error?:string;source?:string}>}){const query=await searchParams;const user=await getSession();if(!user)redirect('/login');const sources=db.prepare('SELECT s.id,s.name,s.base_url,s.sync_status,s.last_synced_at,s.last_error,s.webhook_secret_hash,s.last_webhook_at,s.user_id AS owner_id FROM sources s JOIN source_members sm ON sm.source_id=s.id WHERE sm.user_id=? ORDER BY s.id DESC').all(user.id) as any[];return <>控制台 {query.error&&{query.error}
}{query.source==='shared'?你已加入既有的共享 Memos 來源,不會重複同步貼文。
:query.source&&來源已連接,首次同步已排入佇列。
}發佈到自己的 Memos {sources.length?:請先連接一個 Memos 來源。
} 連接 Memos Token 會使用伺服器金鑰加密保存。同一個 Memos 帳號與網址會自動共用來源,不會建立重複貼文。
已連接來源 {sources.map(s=>{s.name} {s.sync_status}
來源 ID:{s.id} {s.base_url} 上次同步:{s.last_synced_at||'尚未完成'} Webhook:{s.webhook_secret_hash?(s.last_webhook_at?`最近收到:${new Date(s.last_webhook_at+'Z').toLocaleString('zh-TW')}`:'已建立 URL,尚未收到呼叫'):'尚未建立 URL'}{s.last_error&&<>{s.last_error} >}
{s.owner_id===user.id?:這是共享來源;只有建立者可以管理 webhook。
} )} >}
+import { redirect } from "next/navigation";
+import { getSession } from "@/lib/auth";
+import { db } from "@/lib/db";
+import { PublishForm } from "./publish-form";
+import { WebhookControl } from "./webhook-control";
+
+type Source = { id: number; name: string; base_url: string; sync_status: string; last_synced_at: string | null; last_error: string | null; webhook_secret_hash: string | null; last_webhook_at: string | null; owner_id: number; is_enabled: number; disabled_at: string | null };
+type Job = { id: number; kind: string; trigger: string | null; status: string; attempts: number; last_error: string | null; created_at: string; finished_at: string | null };
+
+export const dynamic = "force-dynamic";
+
+export default async function Dashboard({ searchParams }: { searchParams: Promise<{ error?: string; source?: string; sync?: string }> }) {
+ const query = await searchParams; const user = await getSession(); if (!user) redirect("/login");
+ const sourceRows = db.prepare("SELECT s.id,s.name,s.base_url,s.sync_status,s.last_synced_at,s.last_error,s.webhook_secret_hash,s.last_webhook_at,s.user_id AS owner_id,s.is_enabled,s.disabled_at FROM sources s JOIN source_members sm ON sm.source_id=s.id WHERE sm.user_id=? ORDER BY s.id DESC").all(user.id) as Source[];
+ const sources = sourceRows.map((source) => ({ ...source, members: db.prepare("SELECT u.username,u.id,sm.role FROM source_members sm JOIN users u ON u.id=sm.user_id WHERE sm.source_id=? ORDER BY sm.role DESC,u.username").all(source.id) as { username: string; id: number; role: string }[], jobs: db.prepare("SELECT id,kind,trigger,status,attempts,last_error,created_at,finished_at FROM sync_jobs WHERE source_id=? ORDER BY id DESC LIMIT 5").all(source.id) as Job[] }));
+ const publishSources = sources.filter((source) => source.is_enabled);
+ return <>
+ 控制台
+ {query.error && {query.error}
}
+ {query.source === "shared" ? 你已加入既有的共享 Memos 來源,不會重複同步貼文。
: query.source === "updated" ? 來源設定已更新。
: query.source && 來源已連接,首次同步已排入佇列。
}
+ {query.sync === "queued" && 同步已排入佇列。
}{query.sync === "already-queued" && 此來源已有同步工作處理中,不重複排入。
}
+ 發佈到自己的 Memos {publishSources.length ? : 請先連接並啟用一個 Memos 來源。
}
+ 連接 Memos Token 會使用伺服器金鑰加密保存。同一個 Memos 帳號與網址會自動共用來源,不會建立重複貼文。
+ 已連接來源 {sources.map((source) =>
+ {source.name} {source.is_enabled ? source.sync_status : "disabled"}
+ 來源 ID:{source.id} {source.base_url} 成員:{source.members.map((member) => `${member.username}${member.role === "owner" ? "(建立者)" : ""}`).join("、")} 上次同步:{source.last_synced_at || "尚未完成"} Webhook:{source.webhook_secret_hash ? (source.last_webhook_at ? `最近收到:${new Date(source.last_webhook_at + "Z").toLocaleString("zh-TW")}` : "已建立 URL,尚未收到呼叫") : "尚未建立 URL"}{!source.is_enabled && <> 已停用:{source.disabled_at ? new Date(source.disabled_at + "Z").toLocaleString("zh-TW") : "是"}>}{source.last_error && <>{source.last_error} >}
+ {source.owner_id === user.id ? <>
+
+ 來源管理 {source.members.length > 1 && }
+ > : }
+
+ 最近同步工作 {source.jobs.length ? {source.jobs.map((job) => {job.kind} · {job.trigger || "legacy"} · {job.status} · 嘗試 {job.attempts} 次建立:{new Date(job.created_at + "Z").toLocaleString("zh-TW")}{job.finished_at && `;完成:${new Date(job.finished_at + "Z").toLocaleString("zh-TW")}`} {job.last_error && <>{job.last_error} >} )} : 尚無同步工作。
}
+ )}
+ >;
+}
diff --git a/app/layout.tsx b/app/layout.tsx
index 218316b..cd12fcf 100644
--- a/app/layout.tsx
+++ b/app/layout.tsx
@@ -4,5 +4,5 @@ import { getSession } from "@/lib/auth";
export const metadata = { title: "Mebbling", description: "Your Memos hub" };
export default async function RootLayout({ children }: { children: React.ReactNode }) {
const user = await getSession();
- return Mebbling 探索{user ? <> 控制台> : <> 登入 註冊>}{children} ;
+ return Mebbling 探索{user ? <> 控制台 帳號{user.role === "admin" && 管理}> : <> 登入 註冊>}{children} ;
}
diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md
new file mode 100644
index 0000000..9b3d3e9
--- /dev/null
+++ b/docs/OPERATIONS.md
@@ -0,0 +1,32 @@
+# 維運:備份、還原與資料庫升級
+
+## 備份
+
+在專案根目錄執行:
+
+```bash
+./scripts/backup.sh
+```
+
+腳本會在 `data/backups/YYYYMMDD-HHMMSS/` 建立兩個檔案:
+
+- `hub.db`:由正在執行的 SQLite 資料庫建立的一致性備份。
+- `uploads.tar.gz`:Hub 本機上傳的附件。
+
+`data/backups/` 已由 Git 排除。請將備份複製到另一台主機或加密的雲端儲存;只留在同一台機器不算完整備份。
+
+## 還原
+
+1. 停止服務:`docker compose down`。
+2. 備份目前的 `data/hub.db` 與 `public/uploads/`,以免操作失誤。
+3. 將選定備份中的 `hub.db` 覆蓋為 `data/hub.db`。
+4. 解開附件:`tar -xzf data/backups/<時間>/uploads.tar.gz -C public`。
+5. 重新啟動:`docker compose up -d`。
+
+請始終一起還原資料庫與附件,否則貼文中的附件連結可能失效。
+
+## Schema migration
+
+資料庫 schema 由 `lib/db.ts` 管理。每個欄位 migration 在 `schema_migrations` 表中記錄版本與套用時間,啟動 Web 或 Worker 時會自動執行尚未套用的安全 migration。
+
+升級 Mebbling 前請先執行備份。若新版本在測試環境正常運作,再升級正式資料;不支援直接以舊程式碼讀取已升級 schema 的保證。
diff --git a/lib/db.ts b/lib/db.ts
index dfb85f5..2204e6a 100644
--- a/lib/db.ts
+++ b/lib/db.ts
@@ -18,6 +18,7 @@ CREATE TABLE IF NOT EXISTS sources (
id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
name TEXT NOT NULL, base_url TEXT NOT NULL, token_encrypted TEXT NOT NULL, remote_user TEXT,
webhook_supported INTEGER NOT NULL DEFAULT 0, sync_status TEXT NOT NULL DEFAULT 'pending', last_synced_at TEXT, last_error TEXT,
+ is_enabled INTEGER NOT NULL DEFAULT 1, disabled_at TEXT,
created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, UNIQUE(user_id, base_url)
);
CREATE TABLE IF NOT EXISTS posts (
@@ -44,7 +45,8 @@ CREATE TABLE IF NOT EXISTS reports (
CREATE TABLE IF NOT EXISTS sync_jobs (
id INTEGER PRIMARY KEY, source_id INTEGER NOT NULL REFERENCES sources(id) ON DELETE CASCADE,
kind TEXT NOT NULL, payload_json TEXT, status TEXT NOT NULL DEFAULT 'queued', attempts INTEGER NOT NULL DEFAULT 0,
- last_error TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP, run_after TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+ trigger TEXT NOT NULL DEFAULT 'manual', last_error TEXT, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ started_at TEXT, finished_at TEXT, run_after TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS source_members (
source_id INTEGER NOT NULL REFERENCES sources(id) ON DELETE CASCADE,
@@ -55,18 +57,27 @@ CREATE TABLE IF NOT EXISTS source_members (
CREATE INDEX IF NOT EXISTS posts_public_idx ON posts(visibility, hidden, created_at DESC);
CREATE INDEX IF NOT EXISTS sync_jobs_idx ON sync_jobs(status, run_after);
CREATE UNIQUE INDEX IF NOT EXISTS source_remote_identity_unique ON sources(base_url, remote_user) WHERE remote_user IS NOT NULL;
+CREATE TABLE IF NOT EXISTS schema_migrations (
+ version INTEGER PRIMARY KEY, applied_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+);
`);
db.exec("INSERT OR IGNORE INTO source_members(source_id,user_id,role) SELECT id,user_id,'owner' FROM sources");
-const sourceColumns = db.prepare("PRAGMA table_info(sources)").all() as { name: string }[];
-if (!sourceColumns.some((column) => column.name === "webhook_secret_hash")) {
- db.exec("ALTER TABLE sources ADD COLUMN webhook_secret_hash TEXT");
-}
-if (!sourceColumns.some((column) => column.name === "last_webhook_at")) {
- db.exec("ALTER TABLE sources ADD COLUMN last_webhook_at TEXT");
+function applyColumnMigration(version: number, table: string, column: string, sql: string) {
+ const columns = db.prepare(`PRAGMA table_info(${table})`).all() as { name: string }[];
+ if (!columns.some((item) => item.name === column)) db.exec(sql);
+ db.prepare("INSERT OR IGNORE INTO schema_migrations(version) VALUES(?)").run(version);
}
+applyColumnMigration(1, "sources", "webhook_secret_hash", "ALTER TABLE sources ADD COLUMN webhook_secret_hash TEXT");
+applyColumnMigration(2, "sources", "last_webhook_at", "ALTER TABLE sources ADD COLUMN last_webhook_at TEXT");
+applyColumnMigration(3, "sources", "is_enabled", "ALTER TABLE sources ADD COLUMN is_enabled INTEGER NOT NULL DEFAULT 1");
+applyColumnMigration(4, "sources", "disabled_at", "ALTER TABLE sources ADD COLUMN disabled_at TEXT");
+applyColumnMigration(5, "sync_jobs", "trigger", "ALTER TABLE sync_jobs ADD COLUMN trigger TEXT NOT NULL DEFAULT 'manual'");
+applyColumnMigration(6, "sync_jobs", "started_at", "ALTER TABLE sync_jobs ADD COLUMN started_at TEXT");
+applyColumnMigration(7, "sync_jobs", "finished_at", "ALTER TABLE sync_jobs ADD COLUMN finished_at TEXT");
+
const admin = process.env.ADMIN_USERNAME;
const adminPassword = process.env.ADMIN_PASSWORD;
if (admin && adminPassword) {
diff --git a/lib/sync.ts b/lib/sync.ts
new file mode 100644
index 0000000..d31b25c
--- /dev/null
+++ b/lib/sync.ts
@@ -0,0 +1,16 @@
+import { db } from "@/lib/db";
+
+export type SyncTrigger = "manual" | "webhook" | "scheduled" | "source-created";
+
+/** Queue one pull per source at a time. Returns true only when a new job was created. */
+export function queuePull(sourceId: number, trigger: SyncTrigger, payload: unknown = {}) {
+ const result = db.prepare(`
+ INSERT INTO sync_jobs(source_id,kind,payload_json,trigger)
+ SELECT ?, 'pull', ?, ?
+ WHERE NOT EXISTS (
+ SELECT 1 FROM sync_jobs
+ WHERE source_id=? AND kind='pull' AND status IN ('queued','running')
+ )
+ `).run(sourceId, JSON.stringify(payload), trigger, sourceId);
+ return result.changes === 1;
+}
diff --git a/package-lock.json b/package-lock.json
index 41e7eb5..1d886ac 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -1,13 +1,13 @@
{
"name": "mebbling",
- "version": "1.0.0",
+ "version": "0.2.0",
"lockfileVersion": 3,
"requires": true,
"packages": {
"": {
"name": "mebbling",
- "version": "1.0.0",
- "license": "ISC",
+ "version": "0.2.0",
+ "license": "PolyForm-Noncommercial-1.0.0",
"dependencies": {
"bcryptjs": "^3.0.3",
"better-sqlite3": "^12.11.1",
diff --git a/package.json b/package.json
index 8c33253..fa183b9 100644
--- a/package.json
+++ b/package.json
@@ -1,6 +1,6 @@
{
"name": "mebbling",
- "version": "1.0.0",
+ "version": "0.2.0",
"description": "",
"private": true,
"scripts": {
@@ -8,11 +8,11 @@
"build": "HUB_BUILD=1 next build",
"start": "next start",
"worker": "tsx worker/index.ts",
- "test": "tsx --test tests/**/*.test.ts"
+ "test": "TMPDIR=/tmp tsx --test tests/**/*.test.ts"
},
"keywords": [],
"author": "",
- "license": "ISC",
+ "license": "PolyForm-Noncommercial-1.0.0",
"dependencies": {
"bcryptjs": "^3.0.3",
"better-sqlite3": "^12.11.1",
diff --git a/scripts/backup.sh b/scripts/backup.sh
new file mode 100644
index 0000000..2c82928
--- /dev/null
+++ b/scripts/backup.sh
@@ -0,0 +1,18 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Creates a consistent SQLite backup through the running web container, then archives Hub uploads.
+root_dir="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$root_dir"
+stamp="$(date +%Y%m%d-%H%M%S)"
+backup_dir="data/backups/$stamp"
+mkdir -p "$backup_dir"
+
+docker compose exec -T -e BACKUP_PATH="/app/data/backups/$stamp/hub.db" web node -e '
+ const Database = require("better-sqlite3");
+ const db = new Database(process.env.DATABASE_PATH);
+ db.backup(process.env.BACKUP_PATH).then(() => db.close()).catch((error) => { console.error(error); process.exit(1); });
+'
+
+tar -czf "$backup_dir/uploads.tar.gz" -C public uploads
+printf 'Created backup: %s\n' "$backup_dir"
diff --git a/tests/sync.test.ts b/tests/sync.test.ts
new file mode 100644
index 0000000..0097f8f
--- /dev/null
+++ b/tests/sync.test.ts
@@ -0,0 +1,25 @@
+import assert from "node:assert/strict";
+import { after, test } from "node:test";
+import { randomUUID } from "node:crypto";
+import { rmSync } from "node:fs";
+
+const databasePath = `/tmp/mebbling-sync-${randomUUID()}.db`;
+process.env.DATABASE_PATH = databasePath;
+delete process.env.HUB_BUILD;
+
+let database: typeof import("../lib/db").db | undefined;
+
+after(() => { database?.close(); rmSync(databasePath, { force: true }); rmSync(`${databasePath}-wal`, { force: true }); rmSync(`${databasePath}-shm`, { force: true }); });
+
+test("applies tracked migrations and deduplicates active pull jobs", async () => {
+ const { db } = await import("../lib/db"); database = db;
+ const { queuePull } = await import("../lib/sync");
+ const migrations = db.prepare("SELECT version FROM schema_migrations ORDER BY version").all() as { version: number }[];
+ assert.deepEqual(migrations.map((item) => item.version), [1, 2, 3, 4, 5, 6, 7]);
+ const userId = Number(db.prepare("INSERT INTO users(username,password_hash) VALUES('sync-test','hash')").run().lastInsertRowid);
+ const sourceId = Number(db.prepare("INSERT INTO sources(user_id,name,base_url,token_encrypted,is_enabled) VALUES(?,?,?,?,1)").run(userId, "Test", "https://example.test", "encrypted").lastInsertRowid);
+ assert.equal(queuePull(sourceId, "manual"), true);
+ assert.equal(queuePull(sourceId, "webhook", { event: "memo.updated" }), false);
+ const jobs = db.prepare("SELECT kind,trigger,status FROM sync_jobs WHERE source_id=?").all(sourceId) as { kind: string; trigger: string; status: string }[];
+ assert.deepEqual(jobs, [{ kind: "pull", trigger: "manual", status: "queued" }]);
+});
diff --git a/worker/index.ts b/worker/index.ts
index 37b983d..8d42154 100644
--- a/worker/index.ts
+++ b/worker/index.ts
@@ -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();