feat(watchlist): SyncEngine — tiered diff-on-write sync with throttle/backoff
This commit is contained in:
parent
1923bb87b8
commit
eeaa37469c
2 changed files with 218 additions and 0 deletions
95
packages/obsidian/src/watchlist/SyncEngine.ts
Normal file
95
packages/obsidian/src/watchlist/SyncEngine.ts
Normal file
|
|
@ -0,0 +1,95 @@
|
|||
import { parseNote, extractMyNotes, noteTmdbRef } from 'packages/obsidian/src/watchlist/parse';
|
||||
import { buildRecord } from 'packages/obsidian/src/watchlist/build';
|
||||
import { renderNote } from 'packages/obsidian/src/watchlist/render';
|
||||
|
||||
export interface SyncDeps {
|
||||
listNotes(): Promise<{ path: string; content: string }[]>;
|
||||
writeNote(path: string, content: string): Promise<void>;
|
||||
fetchDetail(tmdbId: string, isMovie: boolean): Promise<any>;
|
||||
sleep(ms: number): Promise<void>;
|
||||
log(msg: string): void;
|
||||
}
|
||||
|
||||
export interface SyncOptions {
|
||||
full?: boolean;
|
||||
dryRun?: boolean;
|
||||
throttleMs?: number;
|
||||
}
|
||||
|
||||
export interface SyncReport {
|
||||
scanned: number;
|
||||
synced: number;
|
||||
written: number;
|
||||
skippedNoId: number;
|
||||
skippedStatic: number;
|
||||
flipped: string[];
|
||||
errors: { path: string; error: string }[];
|
||||
}
|
||||
|
||||
export class TmdbRateLimitError extends Error {
|
||||
retryAfterMs: number = 2000;
|
||||
}
|
||||
|
||||
const ACTIVE_STATUSES = new Set(['Returning Series', 'In Production', 'Planned', 'Pilot']);
|
||||
|
||||
function strip(s: string | undefined): string {
|
||||
return (s ?? '').trim().replace(/^"|"$/g, '');
|
||||
}
|
||||
|
||||
export function isActive(fm: Record<string, string>): boolean {
|
||||
const status = strip(fm['status']);
|
||||
if (!status) return true; // never enriched → needs first pass
|
||||
if (ACTIVE_STATUSES.has(status)) return true;
|
||||
if (strip(fm['watch_status']) === 'Watching') return true;
|
||||
const nextAir = strip(fm['next_air_date']);
|
||||
if (nextAir && nextAir !== 'null') return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
export async function syncFolder(deps: SyncDeps, opts: SyncOptions = {}): Promise<SyncReport> {
|
||||
const throttleMs = opts.throttleMs ?? 250;
|
||||
const report: SyncReport = { scanned: 0, synced: 0, written: 0, skippedNoId: 0, skippedStatic: 0, flipped: [], errors: [] };
|
||||
const notes = await deps.listNotes();
|
||||
for (const note of notes) {
|
||||
report.scanned++;
|
||||
const { frontmatter, body } = parseNote(note.content);
|
||||
const ref = noteTmdbRef(frontmatter);
|
||||
if (!ref) {
|
||||
report.skippedNoId++;
|
||||
continue;
|
||||
}
|
||||
if (!opts.full && !isActive(frontmatter)) {
|
||||
report.skippedStatic++;
|
||||
continue;
|
||||
}
|
||||
try {
|
||||
let detail: any;
|
||||
try {
|
||||
detail = await deps.fetchDetail(ref.tmdbId, ref.isMovie);
|
||||
} catch (e) {
|
||||
if (e instanceof TmdbRateLimitError) {
|
||||
await deps.sleep(e.retryAfterMs);
|
||||
detail = await deps.fetchDetail(ref.tmdbId, ref.isMovie);
|
||||
} else {
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
const record = buildRecord(detail, ref.isMovie, frontmatter);
|
||||
const rendered = renderNote(record, extractMyNotes(body));
|
||||
report.synced++;
|
||||
if (strip(frontmatter['watch_status']) === 'Watched' && record.watchStatus === 'Unwatched') {
|
||||
report.flipped.push(note.path);
|
||||
}
|
||||
if (rendered !== note.content) {
|
||||
report.written++;
|
||||
if (!opts.dryRun) await deps.writeNote(note.path, rendered);
|
||||
deps.log(`${opts.dryRun ? '[dry] ' : ''}updated ${note.path}`);
|
||||
}
|
||||
await deps.sleep(throttleMs);
|
||||
} catch (e) {
|
||||
report.errors.push({ path: note.path, error: e instanceof Error ? e.message : String(e) });
|
||||
deps.log(`ERROR ${note.path}: ${String(e)}`);
|
||||
}
|
||||
}
|
||||
return report;
|
||||
}
|
||||
123
tests/watchlist-sync-engine.test.ts
Normal file
123
tests/watchlist-sync-engine.test.ts
Normal file
|
|
@ -0,0 +1,123 @@
|
|||
import { describe, expect, test } from 'bun:test';
|
||||
import { syncFolder, isActive, TmdbRateLimitError, type SyncDeps } from 'packages/obsidian/src/watchlist/SyncEngine';
|
||||
import tvDetail from 'tests/fixtures/tmdb-tv-loki.json';
|
||||
|
||||
const ENDED_NOTE = `---
|
||||
type: watchlist_item
|
||||
media_type: TV Series
|
||||
watch_status: Watched
|
||||
rating: 5
|
||||
rating_stars: ⭐️⭐️⭐️⭐️⭐️
|
||||
status: Ended
|
||||
tmdb_id: 84958
|
||||
last_air_date: 2023-11-09
|
||||
---
|
||||
|
||||
# Loki
|
||||
|
||||
## My Notes
|
||||
|
||||
keep me
|
||||
`;
|
||||
|
||||
const AIRING_NOTE = ENDED_NOTE.replace('status: Ended', 'status: Returning Series').replace('last_air_date: 2023-11-09', 'last_air_date: 2023-10-01');
|
||||
|
||||
function makeDeps(notes: { path: string; content: string }[], detail: any = tvDetail) {
|
||||
const writes: { path: string; content: string }[] = [];
|
||||
const fetches: string[] = [];
|
||||
const deps: SyncDeps = {
|
||||
listNotes: async () => notes,
|
||||
writeNote: async (path, content) => { writes.push({ path, content }); },
|
||||
fetchDetail: async (id) => { fetches.push(id); return detail; },
|
||||
sleep: async () => {},
|
||||
log: () => {},
|
||||
};
|
||||
return { deps, writes, fetches };
|
||||
}
|
||||
|
||||
describe('isActive tiering', () => {
|
||||
test('Returning Series → active', () => expect(isActive({ status: 'Returning Series' })).toBe(true));
|
||||
test('Watching → active regardless of status', () => expect(isActive({ status: 'Ended', watch_status: 'Watching' })).toBe(true));
|
||||
test('next_air_date set → active', () => expect(isActive({ status: 'Ended', next_air_date: '2026-08-01' })).toBe(true));
|
||||
test('Ended → static', () => expect(isActive({ status: 'Ended', next_air_date: 'null' })).toBe(false));
|
||||
test('Released movie → static', () => expect(isActive({ status: 'Released' })).toBe(false));
|
||||
test('missing status → active (needs first enrich)', () => expect(isActive({})).toBe(true));
|
||||
});
|
||||
|
||||
describe('syncFolder', () => {
|
||||
test('default run skips static notes', async () => {
|
||||
const { deps, fetches } = makeDeps([{ path: 'Loki.md', content: ENDED_NOTE }]);
|
||||
const report = await syncFolder(deps);
|
||||
expect(fetches.length).toBe(0);
|
||||
expect(report.skippedStatic).toBe(1);
|
||||
});
|
||||
test('full run processes static notes', async () => {
|
||||
const { deps, fetches } = makeDeps([{ path: 'Loki.md', content: ENDED_NOTE }]);
|
||||
await syncFolder(deps, { full: true });
|
||||
expect(fetches).toEqual(['84958']);
|
||||
});
|
||||
test('diff-on-write: second pass on rendered output writes nothing', async () => {
|
||||
const { deps, writes } = makeDeps([{ path: 'Loki.md', content: ENDED_NOTE }]);
|
||||
await syncFolder(deps, { full: true });
|
||||
const rendered = writes[0].content;
|
||||
const second = makeDeps([{ path: 'Loki.md', content: rendered }]);
|
||||
const report = await syncFolder(second.deps, { full: true });
|
||||
expect(second.writes.length).toBe(0);
|
||||
expect(report.written).toBe(0);
|
||||
});
|
||||
test('watch rule flips through full pipeline', async () => {
|
||||
const { deps, writes } = makeDeps([{ path: 'Loki.md', content: AIRING_NOTE }]);
|
||||
const report = await syncFolder(deps);
|
||||
expect(writes.length).toBe(1);
|
||||
expect(writes[0].content).toContain('watch_status: Unwatched');
|
||||
expect(report.flipped).toEqual(['Loki.md']);
|
||||
});
|
||||
test('My Notes preserved through rewrite', async () => {
|
||||
const { deps, writes } = makeDeps([{ path: 'Loki.md', content: AIRING_NOTE }]);
|
||||
await syncFolder(deps);
|
||||
expect(writes[0].content).toContain('keep me');
|
||||
});
|
||||
test('no tmdb ref → skipped, counted', async () => {
|
||||
const { deps, fetches } = makeDeps([{ path: '_Dashboard.md', content: '# dash' }]);
|
||||
const report = await syncFolder(deps, { full: true });
|
||||
expect(fetches.length).toBe(0);
|
||||
expect(report.skippedNoId).toBe(1);
|
||||
});
|
||||
test('dryRun: no writes, report counts', async () => {
|
||||
const { deps, writes } = makeDeps([{ path: 'Loki.md', content: AIRING_NOTE }]);
|
||||
const report = await syncFolder(deps, { dryRun: true });
|
||||
expect(writes.length).toBe(0);
|
||||
expect(report.written).toBe(1); // counts what WOULD be written
|
||||
});
|
||||
test('429 → sleep(retryAfter) then retry succeeds', async () => {
|
||||
let calls = 0;
|
||||
const slept: number[] = [];
|
||||
const deps: SyncDeps = {
|
||||
listNotes: async () => [{ path: 'Loki.md', content: AIRING_NOTE }],
|
||||
writeNote: async () => {},
|
||||
fetchDetail: async () => {
|
||||
calls++;
|
||||
if (calls === 1) { const e = new TmdbRateLimitError('429'); e.retryAfterMs = 1500; throw e; }
|
||||
return tvDetail;
|
||||
},
|
||||
sleep: async ms => { slept.push(ms); },
|
||||
log: () => {},
|
||||
};
|
||||
const report = await syncFolder(deps);
|
||||
expect(calls).toBe(2);
|
||||
expect(slept).toContain(1500);
|
||||
expect(report.errors.length).toBe(0);
|
||||
});
|
||||
test('fetch error recorded, other notes continue', async () => {
|
||||
const { deps } = makeDeps([
|
||||
{ path: 'Bad.md', content: AIRING_NOTE },
|
||||
{ path: 'Good.md', content: AIRING_NOTE },
|
||||
]);
|
||||
let n = 0;
|
||||
deps.fetchDetail = async () => { n++; if (n === 1) throw new Error('boom'); return tvDetail; };
|
||||
const report = await syncFolder(deps);
|
||||
expect(report.errors.length).toBe(1);
|
||||
expect(report.errors[0].path).toBe('Bad.md');
|
||||
expect(report.synced).toBe(1);
|
||||
});
|
||||
});
|
||||
Loading…
Add table
Add a link
Reference in a new issue