Mebbling/worker/index.ts

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();