koreader-plugin/bookorbit.koplugin/bookorbit_sweep.lua
Rama main · réplica de lectura del repositorio original.
--[[--
Full-library sweep: pushes reading statistics, annotations, book states and
bulk progress for all matched books to BookOrbit. Manual-only; the open book
is covered day to day by the per-book snapshot sync.
The sweep is a module-level singleton driven by small steps chained through
UIManager:scheduleIn so the UI stays responsive. It holds no references to a
ReaderUI/FileManager plugin instance and therefore survives document close.
Per-book sync state advances only after the server acknowledges an upload, so
an interrupted sweep resumes exactly where it left off.
Library-sized phases are sliced: enumeration, ReadHistory resolution and
sidecar extraction run in bounded chunks with a per-callback time budget, each
over a snapshot of its keys that is re-resolved when its chunk actually runs.
Every run carries a generation token, and every scheduled callback verifies it
before touching shared state, which is what makes cancellation safe.
]]
local DocSettings = require("docsettings")
local ReadHistory = require("readhistory")
local UIManager = require("ui/uimanager")
local Trapper = require("ui/trapper")
local ffiutil = require("ffi/util")
local logger = require("logger")
local lfs = require("libs/libkoreader-lfs")
local util = require("util")
local T = ffiutil.template
local _ = require("gettext")
local BookOrbitAnnotations = require("bookorbit_annotations")
local BookOrbitBookmarks = require("bookorbit_bookmarks")
local BookOrbitApi = require("bookorbit_api")
local BookOrbitCapabilities = require("bookorbit_capabilities")
local BookOrbitHighlightSummary = require("bookorbit_highlight_summary")
local BookOrbitQueue = require("bookorbit_queue")
local BookOrbitSidecar = require("bookorbit_sidecar")
local BookOrbitState = require("bookorbit_state")
local BookOrbitStateManager = require("bookorbit_state_manager")
local BookOrbitStatsReader = require("bookorbit_stats_reader")
-- The per-callback time budget is what protects responsiveness, so steps chain
-- with a minimal delay. A fixed sleep between short slices would leave a large
-- sweep idle for most of its wall time.
local STEP_DELAY = 0
-- Only paid after the progress display actually repainted, so e-ink has a
-- moment to settle before the next slice starts.
local REPAINT_DELAY = 0.1
local CHUNK_BUDGET_MS = 20
local CHUNK_ITEMS = 40
-- Sidecar extraction opens a file per book, so it yields more often.
local SIDECAR_CHUNK_ITEMS = 20
local ENUM_PAGE = 50
local MATCH_BATCH = 500
local STATS_BATCH = 500
local ANN_BATCH_BOOKS = 20
local ANN_BATCH_TOTAL = 50
local STATES_BATCH = 200
local PROGRESS_BATCH = 100
local PROGRESS_THROTTLE = 2
local BookOrbitSweep = {
running = false,
}
local active_ctx
local run_generation = 0
local function elapsedMs(started)
local ok, seconds, micros = pcall(ffiutil.gettime)
if ok and type(seconds) == "number" then
return (seconds * 1000 + (micros or 0) / 1000) - started
end
return os.clock() * 1000 - started
end
local function nowMs()
return elapsedMs(0)
end
local function titleFromHistoryItem(item)
local name = item.text
if (not name or name == "") and item.file then
name = item.file:gsub(".*/", "")
end
if not name or name == "" then return nil end
local title = name:gsub("%.[^%.]+$", "")
if title == "" then return name end
return title
end
function BookOrbitSweep.isRunning()
return BookOrbitSweep.running
end
function BookOrbitSweep.syncStatus()
return BookOrbitStateManager.summary()
end
local function isAuthError(err)
return err == 401 or err == 403
end
-- A run only owns shared state while it is the current generation. Anything
-- scheduled by a cancelled or superseded run stops here instead of writing.
local function aborted(ctx)
return ctx.finished or ctx.cancelled or ctx.generation ~= run_generation
end
local function closeStatsSession(ctx)
if ctx.stats_session then
ctx.stats_session:close()
ctx.stats_session = nil
end
end
-- The sync sheet stays up at the end to show the result; this only drops
-- the run's hold on it.
local function closeProgress(ctx)
ctx.progress_sheet_done = true
end
-- Interactive-only progress display, reused across phases. Identical text never
-- redraws, and changing text redraws at most every PROGRESS_THROTTLE seconds,
-- so a long first run stays legible without flooding the e-ink screen. The
-- dialog carries the cancel affordance and is not tap-dismissable, so a stray
-- tap cannot silently drop the only way to stop the run.
-- short is the few words a BookOrbit page shows in place of the dialog.
local function setProgress(ctx, text, force, short, done, total)
if not ctx.interactive then return end
if not force then
if text == ctx.progress_shown_text then return end
if (ctx.sync_sheet or ctx.quiet) and os.time() - (ctx.progress_shown_at or 0) < PROGRESS_THROTTLE then
return
end
end
ctx.progress_shown_text = text
ctx.progress_shown_at = os.time()
ctx.progress_repainted = true
-- Started from a BookOrbit page: the page shows progress (and the way
-- to stop) where the sync was asked for, instead of a sheet over it.
if ctx.quiet then
if ctx.on_progress then pcall(ctx.on_progress, short or text) end
return
end
if not ctx.sync_sheet then
ctx.sync_sheet = require("bookorbit_sync_sheet").open{
recheck = ctx.full_recheck,
on_stop = function() BookOrbitSweep.cancel("user") end,
}
end
ctx.sync_sheet:progressed(short or text, done, total)
end
-- How the run ended: in the sync sheet when one is up, else as a pill.
local function report(ctx, result)
if ctx.sync_sheet and not ctx.sync_sheet.closed then
ctx.sync_sheet:finished(result)
return
end
local Toast = require("bookorbit_toast")
if result.error then Toast.error(result.title) else Toast.show(result.title) end
end
local function chunkSummary(ctx)
return string.format(
"enumerateMs=%.0f partialMd5Ms=%.0f partialMd5Count=%d sidecarMs=%.0f slowestItemMs=%.0f statsSkipped=%d ambiguousStats=%d",
ctx.timing.enumerate_ms, ctx.timing.partial_md5_ms, ctx.timing.partial_md5_count,
ctx.timing.sidecar_ms, ctx.timing.max_item_ms, ctx.counts.stats_skipped or 0,
ctx.counts.stats_ambiguous or 0)
end
local function finish(ctx, err)
if ctx.finished then return end
ctx.finished = true
BookOrbitSweep.running = false
if active_ctx == ctx then active_ctx = nil end
closeStatsSession(ctx)
closeProgress(ctx)
-- A failing flush must not re-enter finish through the caller's error
-- handler; the counts below still describe what the server accepted.
local flushed, flush_err = pcall(ctx.state.flush, ctx.state)
if not flushed then
logger.err("BookOrbit: sweep state flush failed:", flush_err)
ctx.had_errors = true
end
if err == "cancelled" then
if ctx.interactive and not ctx.quiet then
report(ctx, { error = true, title = _("Sync stopped"), sub = _("What synced so far is kept. The rest syncs next time.") })
end
logger.info(string.format(
"BookOrbit: sweep cancelled reason=%s matched=%d pageStats=%d annotations=%d %s",
tostring(ctx.cancel_reason or "user"), ctx.counts.books_matched, ctx.counts.page_stats,
ctx.counts.annotations, chunkSummary(ctx)))
if ctx.on_finish then
pcall(ctx.on_finish, err)
end
return
end
if ctx.plugin and ctx.plugin.recordSyncError then
if err == "auth" or err == "network" or err == "server" then
ctx.plugin:recordSyncError("sweep", err)
elseif not err and ctx.had_errors then
ctx.plugin:recordSyncError("sweep", "partial_failure")
elseif not err and ctx.plugin.recordSyncSuccess then
local message = T(_("%1 books, %2 reading events, %3 highlights"),
ctx.counts.books_matched, ctx.counts.page_stats, ctx.counts.annotations)
if (ctx.counts.stats_ambiguous or 0) > 0 then
message = message .. ". " .. T(_("%1 books need to be opened and synced once to recover their reading history."),
ctx.counts.stats_ambiguous)
end
ctx.plugin:recordSyncSuccess(
"sweep",
message)
end
end
if ctx.plugin and ctx.plugin.recordHighlightSync and ctx.highlight_summary
and BookOrbitHighlightSummary.hasCounts(ctx.highlight_summary) then
local highlight_err = ctx.highlight_unsupported and "unsupported_server" or nil
if ctx.had_errors and not highlight_err then
highlight_err = "partial_failure"
end
ctx.plugin:recordHighlightSync(ctx.highlight_summary, highlight_err)
end
if err == "auth" then
if ctx.interactive and not ctx.quiet then
report(ctx, { error = true, title = _("Sign-in failed"), sub = _("Check your KOReader login in BookOrbit: Settings, KOReader.") })
elseif not ctx.quiet then
require("bookorbit_toast").error(_("BookOrbit sign-in failed"))
end
if ctx.on_finish then
pcall(ctx.on_finish, err)
end
return
end
if err == "network" then
if ctx.interactive and not ctx.quiet then
report(ctx, { error = true, title = _("BookOrbit is not reachable"), sub = _("Check the connection and try again.") })
end
logger.dbg("BookOrbit: sweep aborted, server not reachable")
if ctx.on_finish then
pcall(ctx.on_finish, err)
end
return
end
if err == "server" then
if ctx.interactive and not ctx.quiet then
report(ctx, { error = true, title = _("The server could not finish the sync"), sub = _("Try again in a moment.") })
end
logger.warn("BookOrbit: sweep aborted, server request failed")
if ctx.on_finish then
pcall(ctx.on_finish, err)
end
return
end
if ctx.interactive and not ctx.quiet then
local notes = {}
if ctx.highlight_summary and BookOrbitHighlightSummary.hasRemoteChanges(ctx.highlight_summary) then
table.insert(notes, T(_("Highlights updated: %1 applied, %2 deleted, %3 closed book(s)."),
ctx.highlight_summary.applied, ctx.highlight_summary.deleted, ctx.highlight_summary.touched_books))
end
if ctx.highlight_unsupported then
table.insert(notes, _("BookOrbit server needs an update for two-way highlights."))
end
if ctx.highlight_disabled_reported then
table.insert(notes, _("Two-way highlight and bookmark sync is disabled."))
end
if ctx.had_errors then
table.insert(notes, _("Some books failed and will retry on the next sync."))
end
if (ctx.counts.stats_ambiguous or 0) > 0 then
table.insert(notes, T(_("%1 books need to be opened and synced once to recover their reading history."),
ctx.counts.stats_ambiguous))
end
report(ctx, {
title = ctx.full_recheck and _("Books rechecked") or _("Library synced"),
stats = {
{ value = ctx.counts.books_matched, label = _("books matched") },
{ value = ctx.counts.page_stats, label = _("reading events") },
{ value = ctx.counts.annotations, label = _("highlights") },
},
notes = notes,
})
end
logger.info(string.format(
"BookOrbit: sweep done matched=%d pageStats=%d annotations=%d applied=%d deleted=%d %s errors=%s",
ctx.counts.books_matched, ctx.counts.page_stats, ctx.counts.annotations,
ctx.counts.ann_applied or 0, ctx.counts.ann_deleted or 0, chunkSummary(ctx),
tostring(ctx.had_errors)))
if ctx.on_finish then
pcall(ctx.on_finish, err)
end
end
-- Stops the active run at its next yield. No rollback is needed: per-book sync
-- state advances only after the server acknowledged an upload, so a cancelled
-- run just stops scheduling, releases the statistics session and reports
-- itself as cancelled rather than failed.
function BookOrbitSweep.cancel(reason)
local ctx = active_ctx
if not ctx or ctx.finished then return false end
ctx.cancelled = true
ctx.cancel_reason = reason or "user"
-- Bumping the generation is what disowns callbacks that are already
-- scheduled and results that arrive from work still in flight.
run_generation = run_generation + 1
finish(ctx, "cancelled")
return true
end
local function nextDelay(ctx)
if ctx.progress_repainted then
ctx.progress_repainted = false
return REPAINT_DELAY
end
return STEP_DELAY
end
local function guardedRun(ctx, fn)
if aborted(ctx) then return end
local ok, err = pcall(fn, ctx)
if not ok then
logger.err("BookOrbit: sweep step failed:", err)
ctx.had_errors = true
finish(ctx)
end
end
-- Phase transition that may issue requests, so it runs inside Trapper.
local function step(ctx, fn)
UIManager:scheduleIn(nextDelay(ctx), function()
Trapper:wrap(function()
guardedRun(ctx, fn)
end)
end)
end
-- Continuation of a CPU or filesystem slice. No request can originate here, so
-- it skips the Trapper coroutine.
local function chunk(ctx, fn)
UIManager:scheduleIn(nextDelay(ctx), function()
guardedRun(ctx, fn)
end)
end
--[[--
Drives `handler` over a snapshot of keys, yielding after CHUNK_ITEMS items or
CHUNK_BUDGET_MS of work, whichever comes first.
Keys are snapshotted at phase start and re-resolved when their chunk runs: once
a phase yields the user can open a book, which reorders ReadHistory and mutates
plugin state, so index-based iteration over a live list would skip or
double-process entries. Handlers tolerate a key that no longer resolves;
activity that appears mid-sweep belongs to the next run.
A time check after an item cannot preempt that item's filesystem calls, so the
slowest single item is recorded separately rather than assumed to be bounded.
]]
local function forEachChunk(ctx, keys, handler, opts, on_complete)
opts = opts or {}
local limit = opts.items or CHUNK_ITEMS
local total = #keys
local index = 1
local function pump()
if opts.on_chunk_start then opts.on_chunk_start(ctx) end
local started = nowMs()
local processed = 0
while index <= total do
local item_started = nowMs()
handler(ctx, keys[index], index)
local item_ms = elapsedMs(item_started)
if item_ms > ctx.timing.max_item_ms then ctx.timing.max_item_ms = item_ms end
index = index + 1
processed = processed + 1
if processed >= limit or elapsedMs(started) >= CHUNK_BUDGET_MS then break end
end
if opts.elapsed_key then
ctx.timing[opts.elapsed_key] = ctx.timing[opts.elapsed_key] + elapsedMs(started)
end
if index > total then
return on_complete()
end
if opts.progress then opts.progress(ctx, index - 1, total) end
chunk(ctx, pump)
end
pump()
end
-- Phase 1a: enumerate candidate books from the statistics database (primary
-- source, has md5 directly). Keyset pagination keeps each page's query cheap
-- and lets input run between pages of a 10,000-book library.
local function stepEnumerateStats(ctx)
ctx.candidates = {}
ctx.collector = BookOrbitStatsReader.newBookCollector()
if not ctx.stats_session then
return step(ctx, ctx.steps.enumerateCandidates)
end
local after_id = 0
local rows_read = 0
local function pump()
local started = nowMs()
local rows, last_id = ctx.stats_session:bookRowsAfter(after_id, ENUM_PAGE)
for _, row in ipairs(rows) do
BookOrbitStatsReader.collectBookRow(ctx.collector, row)
end
ctx.timing.enumerate_ms = ctx.timing.enumerate_ms + elapsedMs(started)
rows_read = rows_read + #rows
after_id = last_id
if #rows < ENUM_PAGE then
return step(ctx, ctx.steps.enumerateCandidates)
end
setProgress(ctx, T(_("BookOrbit sync: reading local library (%1)"), rows_read), nil, _("Reading library"))
chunk(ctx, pump)
end
pump()
end
-- Phase 1b: turn the collected statistics rows into candidates. Grouping is
-- only complete once every page landed, so ambiguity is resolved here.
local function stepEnumerateCandidates(ctx)
local entries = BookOrbitStatsReader.collectedEntries(ctx.collector)
local function convert(chunk_ctx, entry)
BookOrbitStatsReader.finalizeBookEntry(entry)
chunk_ctx.candidates[entry.md5] = {
stat_ids = entry.ids,
stats_rows = entry.rows,
last_open = entry.last_open,
title = entry.title,
authors = entry.authors,
source = "statistics",
metadata_ambiguous = entry.metadata_ambiguous,
stats_metadata_ambiguous = entry.stats_ambiguous,
}
end
forEachChunk(ctx, entries, convert, { items = 200, elapsed_key = "enumerate_ms" }, function()
ctx.collector = nil
step(ctx, ctx.steps.enumerateHistory)
end)
end
-- Phase 1c: resolve ReadHistory entries, which contribute file paths for
-- sidecar data. Uncached digests are sampled here, which is the one part of
-- enumeration that touches book files.
local function historyEntry(ctx, file)
local item = ctx.history_lookup[file]
if not item then return end
local file_exists = lfs.attributes(file, "mode") == "file"
local md5 = ctx.state.files[file]
local mapped_book = md5 and ctx.state:getBook(md5) or nil
local identity_conflict = mapped_book and mapped_book.file and mapped_book.file ~= file
if identity_conflict and file_exists then
local started = nowMs()
local ok, computed = pcall(util.partialMD5, file)
ctx.timing.partial_md5_ms = ctx.timing.partial_md5_ms + elapsedMs(started)
ctx.timing.partial_md5_count = ctx.timing.partial_md5_count + 1
if ok and computed then
if computed ~= md5 then
ctx.state:repairFileIdentity(file, md5, computed)
end
md5 = computed
end
end
if not md5 and file_exists and DocSettings:hasSidecarFile(file) then
local doc_settings = DocSettings:open(file)
md5 = doc_settings:readSetting("partial_md5_checksum")
if not md5 then
local started = nowMs()
local ok, computed = pcall(util.partialMD5, file)
ctx.timing.partial_md5_ms = ctx.timing.partial_md5_ms + elapsedMs(started)
ctx.timing.partial_md5_count = ctx.timing.partial_md5_count + 1
if ok then md5 = computed end
end
ctx.state:rememberFile(file, md5)
end
if not md5 then return end
local cand = ctx.candidates[md5] or {}
if file_exists and (not cand.file or (item.time or 0) > (cand.file_last_open or -1)) then
cand.file = file
cand.file_last_open = item.time or 0
cand.source = "file"
if cand.metadata_ambiguous then
cand.title = titleFromHistoryItem(item)
cand.authors = nil
elseif not cand.title then
cand.title = titleFromHistoryItem(item)
end
cand.metadata_ambiguous = false
end
if (item.time or 0) > (cand.last_open or 0) then
cand.last_open = item.time
end
ctx.candidates[md5] = cand
local book = ctx.state:getBook(md5)
if book and file_exists then
cand.book_file_id = book.fileId
end
if book and file_exists and not book.file then
book.file = file
end
if book and cand.file and book.file ~= cand.file then
book.file = cand.file
book.sidecarMtime = nil
end
end
-- Opening a book reorders ReadHistory, so the live list is re-indexed by file
-- at the start of every chunk rather than trusted across a yield.
local function rebuildHistoryLookup(ctx)
local lookup = {}
for _, item in ipairs(ReadHistory.hist or {}) do
if item.file then lookup[item.file] = item end
end
ctx.history_lookup = lookup
end
local function stepEnumerateHistory(ctx)
local files = {}
for _, item in ipairs(ReadHistory.hist or {}) do
if item.file then table.insert(files, item.file) end
end
forEachChunk(ctx, files, historyEntry, {
on_chunk_start = rebuildHistoryLookup,
elapsed_key = "enumerate_ms",
}, function()
ctx.history_lookup = nil
step(ctx, ctx.steps.match)
end)
end
-- Phase 2: ask the server which hashes it knows. A normal run checks only what
-- it cannot answer locally: never-seen hashes, unmatched hashes with new
-- activity, and matched books whose freshness stamp is missing or expired.
-- A full recheck rechecks every known hash instead.
local function stepMatch(ctx)
local to_check = {}
local queued = {}
local now = os.time()
local function queue(md5)
if not queued[md5] then
queued[md5] = true
table.insert(to_check, md5)
end
end
for md5, cand in pairs(ctx.candidates) do
if ctx.full_recheck then
queue(md5)
else
local book = ctx.state:getBook(md5)
if not book then
local last_check = ctx.state.unmatched[md5]
if not last_check or (cand.last_open or 0) > last_check then
queue(md5)
end
elseif not BookOrbitState.isMatchFresh(book, ctx.state.global, now) then
-- The batched refresh path for stale stamps. Per-book syncs
-- rely on it, which is why an expired stamp has to land in a
-- normal sweep and not only in an explicit recheck.
queue(md5)
end
end
end
if ctx.full_recheck then
for md5 in pairs(ctx.state.books) do
queue(md5)
end
for md5 in pairs(ctx.state.unmatched) do
queue(md5)
end
else
-- A matched book can outlive both its statistics row and its history
-- entry, so its stamp still expires with no candidate to carry it.
for md5, book in pairs(ctx.state.books) do
if not ctx.candidates[md5] and not BookOrbitState.isMatchFresh(book, ctx.state.global, now) then
queue(md5)
end
end
end
ctx.match_queue = BookOrbitQueue.new()
for i = 1, #to_check, MATCH_BATCH do
local batch = {}
for j = i, math.min(i + MATCH_BATCH - 1, #to_check) do
table.insert(batch, to_check[j])
end
ctx.match_queue:push(batch)
end
step(ctx, ctx.steps.matchNext)
end
local function hasNewEvents(latest, ids, watermark)
for _, id in ipairs(ids) do
if (latest[id] or 0) > watermark then return true end
end
return false
end
local function buildStatsQueue(ctx)
-- One grouped query answers "does this book have anything after its
-- watermark" for the whole library, so unchanged books cost no per-book
-- event query at all. A failed query returns nil and everything is queued.
local latest = ctx.stats_session and ctx.stats_session:latestEventTimes() or nil
ctx.stats_queue = BookOrbitQueue.new()
ctx.counts.stats_skipped = 0
ctx.counts.stats_ambiguous = 0
for md5, cand in pairs(ctx.candidates) do
local book = ctx.state:getBook(md5)
if cand.stat_ids and book then
if cand.stats_metadata_ambiguous then
for _, row in ipairs(cand.stats_rows or {}) do
local target = ctx.state:getStatsRow(row.id, md5, row.title, row.authors)
if target.bookFileId then
if latest and not hasNewEvents(latest, { row.id }, target.statsWatermark or 0) then
ctx.counts.stats_skipped = ctx.counts.stats_skipped + 1
else
ctx.stats_queue:push({
md5 = md5,
ids = { row.id },
target = target,
book_file_id = target.bookFileId,
})
end
elseif not latest or (latest[row.id] or 0) > (target.statsWatermark or 0) then
ctx.counts.stats_ambiguous = ctx.counts.stats_ambiguous + 1
end
end
else
if latest and not hasNewEvents(latest, cand.stat_ids, book.statsWatermark or 0) then
ctx.counts.stats_skipped = ctx.counts.stats_skipped + 1
else
ctx.stats_queue:push({ md5 = md5, ids = cand.stat_ids })
end
end
end
end
end
local function stepMatchNext(ctx)
local batch = ctx.match_queue:pop()
if not batch then
for md5 in pairs(ctx.candidates) do
if ctx.state:getBook(md5) then
ctx.counts.books_matched = ctx.counts.books_matched + 1
end
end
ctx.state:flush()
buildStatsQueue(ctx)
step(ctx, ctx.steps.statsNext)
return
end
setProgress(ctx, _("BookOrbit sync: matching books..."), nil, _("Matching books"))
local body, err = ctx.client:matchCheck(batch, ctx.candidates)
if aborted(ctx) then return end
if not body then
if isAuthError(err) then return finish(ctx, "auth") end
if type(err) == "number" then return finish(ctx, "server") end
return finish(ctx, "network")
end
local matched = {}
for _, match in ipairs(body.matches or {}) do
matched[match.hash] = true
local cand = ctx.candidates[match.hash]
-- Stamped against the token this response carried, not the one still
-- stored, so these books stay on the fast path after the sweep records
-- a changed library version.
ctx.state:setMatched(match.hash, match.bookFileId, match.bookId,
cand and cand.file or nil, body.libraryVersion, match.positionCompatible)
end
for _, md5 in ipairs(batch) do
if not matched[md5] then
ctx.state:setUnmatched(md5)
end
end
if body.libraryVersion then
ctx.server_library_version = body.libraryVersion
end
step(ctx, ctx.steps.matchNext)
end
-- Phase 3: upload page stat events per matched book, batched and watermarked.
local function stepStatsNext(ctx)
local item = ctx.current_stats or ctx.stats_queue:pop()
ctx.current_stats = nil
if not item then
step(ctx, ctx.steps.sidecars)
return
end
local total = ctx.stats_queue:total()
if total > 0 then
setProgress(ctx, T(_("BookOrbit sync: uploading reading data (%1/%2)"),
ctx.stats_queue:done(), total), nil, T(_("Reading data %1/%2"), ctx.stats_queue:done(), total), ctx.stats_queue:done(), total)
end
local book = ctx.state:getBook(item.md5)
if not book then
step(ctx, ctx.steps.statsNext)
return
end
local stats_target = item.target or book
local watermark = stats_target.statsWatermark or 0
local events = ctx.stats_session
and ctx.stats_session:eventsAfter(item.ids, watermark, STATS_BATCH)
or BookOrbitStatsReader.getEventsAfter(item.ids, watermark, STATS_BATCH)
if not events or #events == 0 then
step(ctx, ctx.steps.statsNext)
return
end
local upload = { hash = item.md5, events = events }
if item.book_file_id then upload.bookFileId = item.book_file_id end
local body, err = ctx.client:uploadPageStats({ upload })
if aborted(ctx) then return end
if not body then
if isAuthError(err) then return finish(ctx, "auth") end
logger.dbg("BookOrbit: page stats upload failed for", item.md5, err)
ctx.had_errors = true
step(ctx, ctx.steps.statsNext)
return
end
for _, unmatched in ipairs(body.unmatched or {}) do
if unmatched == item.md5 then
if item.book_file_id then
stats_target.bookFileId = nil
stats_target.bookId = nil
stats_target.statsWatermark = 0
ctx.had_errors = true
else
ctx.state:setUnmatched(item.md5)
end
step(ctx, ctx.steps.statsNext)
return
end
end
local more = BookOrbitState.applyStatsAck(stats_target, events, body, item.md5, STATS_BATCH, watermark)
ctx.counts.page_stats = ctx.counts.page_stats + #events
if more then
ctx.current_stats = item
end
step(ctx, ctx.steps.statsNext)
end
local function currentlyOpenFile()
local ok, ReaderUI = pcall(require, "apps/reader/readerui")
if ok and ReaderUI and ReaderUI.instance and ReaderUI.instance.document then
return ReaderUI.instance.document.file
end
end
local PENDING_CHUNK = 500
-- Only books whose sidecars changed are exchanged, so a highlight made
-- elsewhere would never reach a closed book nobody opened. The server names
-- the books with something waiting for this device; they join the queue. The
-- open book is left out: its changes are applied live, not by the sweep. So is
-- a book never opened here: it has no sidecar, and its first open delivers.
local function fetchServerPending(ctx)
ctx.server_pending = {}
if not ctx.annotation_sync or ctx.use_legacy_annotations then return end
if BookOrbitCapabilities.supports(ctx.client, "annotationPendingBooks") ~= true then return end
local open_file = currentlyOpenFile()
local hashes = {}
for md5, book in pairs(ctx.state.books) do
if book.file and book.file ~= open_file and BookOrbitSidecar.sidecarMtime(book.file) then
table.insert(hashes, md5)
end
end
for start = 1, #hashes, PENDING_CHUNK do
local chunk = {}
for index = start, math.min(start + PENDING_CHUNK - 1, #hashes) do
table.insert(chunk, hashes[index])
end
local body, err = ctx.client:annotationPendingBooks(chunk)
if not body then
if err == 404 then BookOrbitCapabilities.markUnsupported(ctx.client, "annotationPendingBooks") end
logger.dbg("BookOrbit: pending annotation books unavailable:", err)
return
end
for _, md5 in ipairs(body.pending or {}) do
ctx.server_pending[md5] = true
end
end
end
-- Phase 4: read sidecars of matched books with known paths, mtime-gated, and
-- queue annotation/state/progress deltas. Each book opens a sidecar file, so
-- this runs in small chunks over a snapshot of the matched digests.
local function sidecarEntry(ctx, md5)
local book = ctx.state:getBook(md5)
if not book or not book.file then return end
local mtime = BookOrbitSidecar.sidecarMtime(book.file)
if not mtime then return end
local sidecar_changed = mtime ~= book.sidecarMtime or (ctx.server_pending and ctx.server_pending[md5]) == true
local state_unknown = book.statusSyncedKnown ~= true or book.ratingSyncedKnown ~= true or book.reviewSyncedKnown ~= true
local extract = nil
if sidecar_changed or state_unknown then
extract = BookOrbitSidecar.extract(book.file)
end
if not extract then
if not sidecar_changed then
local state_payload = { hash = md5 }
ctx.pending_books[md5] = {
md5 = md5,
mtime = mtime,
ann_count = book.annCount or 0,
ann_done = true,
failed = false,
need_state = true,
need_progress = false,
state_payload = state_payload,
state_summary = {},
}
ctx.states_items:push(state_payload)
end
return
end
local pending = {
md5 = md5,
mtime = mtime,
ann_count = sidecar_changed and extract.annotations_count or (book.annCount or 0),
ann_max_datetime = sidecar_changed and extract.annotations_max_datetime or nil,
ann_signature = sidecar_changed and extract.annotations_signature or nil,
ann_done = not sidecar_changed,
failed = false,
need_state = false,
need_progress = false,
}
if sidecar_changed then
-- Every changed sidecar goes through the exchange, even with no new
-- local highlights: the full key set is what lets the server detect
-- on-device deletions.
ctx.ann_queue:push({
md5 = md5,
file = book.file,
-- When the sidecar was written: the open book's copy in memory can be newer.
captured_at = mtime,
annotations = extract.annotations,
ann_max_datetime = extract.annotations_max_datetime,
ann_signature = extract.annotations_signature,
bookmarks = extract.bookmarks,
bm_signature = extract.bookmarks_signature,
sidecar_marked = extract.marked,
sidecar_state = BookOrbitSidecar.sidecarState(extract.marked,
(extract.annotations_count or 0) + (extract.bookmarks_count or 0)),
})
end
local state_payload = BookOrbitSidecar.buildStatePayload(md5, book, {
status = extract.status,
status_modified = extract.status_modified,
rating = extract.rating,
review_note = extract.review_note,
}, true)
if state_payload then
pending.need_state = true
pending.status_modified = state_payload.statusModified
pending.state_payload = state_payload
pending.state_summary = {
rating = extract.rating,
review_note = extract.review_note,
}
ctx.states_items:push(state_payload)
end
if sidecar_changed and extract.percent_finished then
local pushed = book.progressPushedPct or -1
if math.abs(extract.percent_finished - pushed) > 0.001 then
pending.need_progress = true
pending.percentage = extract.percent_finished
local cand = ctx.candidates[md5]
ctx.progress_items:push({
hash = md5,
percentage = extract.percent_finished,
progress = extract.last_position,
timestamp = cand and cand.last_open or nil,
})
end
end
ctx.pending_books[md5] = pending
end
local function stepSidecars(ctx)
fetchServerPending(ctx)
ctx.pending_books = {}
ctx.ann_queue = BookOrbitQueue.new()
ctx.ann_chunks = BookOrbitQueue.new()
ctx.states_items = BookOrbitQueue.new()
ctx.progress_items = BookOrbitQueue.new()
local digests = {}
for md5 in pairs(ctx.state.books) do
table.insert(digests, md5)
end
forEachChunk(ctx, digests, sidecarEntry, {
items = SIDECAR_CHUNK_ITEMS,
elapsed_key = "sidecar_ms",
progress = function(progress_ctx, done, total)
setProgress(progress_ctx, T(_("BookOrbit sync: checking books (%1/%2)"), done, total), nil, T(_("Checking %1/%2"), done, total), done, total)
end,
}, function()
step(ctx, ctx.steps.annotationsNext)
end)
end
local buildLegacyAnnotationChunks
-- Dogears share the sidecar and the apply mode with highlights, so the sweep
-- exchanges both for the same book in one pass. A server without the route is
-- recorded once and skipped for the rest of the run.
-- Returns true when the dogears are settled for this pass (exchanged cleanly,
-- skipped as unchanged, or not synced at all), which the sidecar marker needs.
local function exchangeBookmarksForEntry(ctx, entry, apply_mode, pending)
if not ctx.annotation_sync or ctx.bookmarks_unsupported then return true end
if not BookOrbitBookmarks.enabled(ctx.client, true) then
ctx.bookmarks_unsupported = true
return true
end
local book = ctx.state:getBook(entry.md5)
if not book then return false end
if not ctx.interactive and BookOrbitBookmarks.canSkipExchange(book, entry.bm_signature) then return true end
local result, err = BookOrbitBookmarks.exchangeBook({
client = ctx.client,
state = ctx.state,
digest = entry.md5,
bookmarks = entry.bookmarks,
bm_signature = entry.bm_signature,
sidecar_state = entry.sidecar_state,
apply_mode = apply_mode,
file = entry.file,
})
if not result then
if err == "unsupported_server" then
BookOrbitBookmarks.markUnsupported(ctx.client)
ctx.bookmarks_unsupported = true
return true
end
ctx.had_errors = true
if pending then pending.failed = true end
ctx.highlight_summary = BookOrbitHighlightSummary.addBookmarks(ctx.highlight_summary, { had_errors = true })
logger.dbg("BookOrbit: bookmark exchange failed for", entry.md5, err)
return false
end
if result.had_errors then
ctx.had_errors = true
if pending then pending.failed = true end
end
ctx.counts.bookmarks = (ctx.counts.bookmarks or 0) + result.uploaded
ctx.highlight_summary = BookOrbitHighlightSummary.addBookmarks(ctx.highlight_summary, result)
return not result.had_errors
end
-- Phase 5: per-book annotation exchange. Uploads the local delta, reports the
-- key set for deletion detection and applies server-side changes into the
-- sidecar of closed books. Falls back to the legacy one-way upload when the
-- server has no exchange endpoint.
local function stepAnnotationsNext(ctx)
if ctx.use_legacy_annotations then
return step(ctx, ctx.steps.annotationsLegacyNext)
end
local entry = ctx.ann_queue:pop()
if not entry then
step(ctx, ctx.steps.statesNext)
return
end
local total = ctx.ann_queue:total()
if total > 0 then
setProgress(ctx, T(_("BookOrbit sync: syncing highlights (%1/%2)"),
ctx.ann_queue:done(), total), nil, T(_("Highlights %1/%2"), ctx.ann_queue:done(), total), ctx.ann_queue:done(), total)
end
local pending = ctx.pending_books[entry.md5]
local apply_mode = "skip"
if ctx.annotation_sync and entry.file ~= currentlyOpenFile() then
apply_mode = "sidecar"
elseif not ctx.annotation_sync and ctx.interactive and not ctx.highlight_disabled_reported then
ctx.highlight_disabled_reported = true
ctx.highlight_summary = BookOrbitHighlightSummary.add(ctx.highlight_summary, nil, { skipped = 1 })
end
local result, err = BookOrbitAnnotations.exchangeBook({
client = ctx.client,
state = ctx.state,
digest = entry.md5,
annotations = entry.annotations,
ann_max_datetime = entry.ann_max_datetime,
ann_signature = entry.ann_signature,
captured_at = entry.captured_at,
sidecar_state = entry.sidecar_state,
apply_mode = apply_mode,
file = entry.file,
})
if aborted(ctx) then return end
if not result then
if isAuthError(err) or err == "auth" then return finish(ctx, "auth") end
if err == "unsupported_server" then
-- Re-queue everything, including this entry, for the legacy path.
ctx.ann_queue:requeue(entry)
ctx.use_legacy_annotations = true
ctx.highlight_unsupported = true
ctx.highlight_summary = BookOrbitHighlightSummary.add(ctx.highlight_summary, nil, { skipped = 1 })
buildLegacyAnnotationChunks(ctx)
return step(ctx, ctx.steps.annotationsLegacyNext)
end
ctx.had_errors = true
if pending then pending.failed = true end
ctx.highlight_summary = BookOrbitHighlightSummary.add(ctx.highlight_summary, { had_errors = true })
logger.dbg("BookOrbit: annotation exchange failed for", entry.md5, err)
return step(ctx, ctx.steps.annotationsNext)
end
if result.had_errors then
ctx.had_errors = true
if pending then pending.failed = true end
elseif pending then
pending.ann_done = true
end
ctx.counts.annotations = ctx.counts.annotations + result.uploaded
ctx.counts.ann_applied = (ctx.counts.ann_applied or 0) + result.applied
ctx.counts.ann_deleted = (ctx.counts.ann_deleted or 0) + result.deleted
ctx.highlight_summary = BookOrbitHighlightSummary.add(ctx.highlight_summary, result, {
closed_book = apply_mode == "sidecar",
})
local bookmarks_done = exchangeBookmarksForEntry(ctx, entry, apply_mode, pending)
if aborted(ctx) then return end
if bookmarks_done and not result.had_errors and not result.incompatible
and apply_mode == "sidecar" and entry.sidecar_marked == false
and BookOrbitSidecar.markFile(entry.file) and pending then
-- The marker is this sweep's own write; without this the next sweep would
-- see a changed sidecar and exchange the book again for nothing.
pending.mtime = BookOrbitSidecar.sidecarMtime(entry.file) or pending.mtime
end
return step(ctx, ctx.steps.annotationsNext)
end
-- Legacy phase 5: one-way chunk upload, packing several books per request
-- while respecting both the per-request book and annotation caps. It walks the
-- queue from the head, because index 1 is a consumed slot by this point.
buildLegacyAnnotationChunks = function(ctx)
local device_now = os.date("%Y-%m-%d %H:%M:%S")
for entry in ctx.ann_queue:iter() do
local book = ctx.state:getBook(entry.md5)
local ann_watermark = book and BookOrbitAnnotations.readWatermark(book, device_now) or ""
local delta = {}
for _, annotation in ipairs(entry.annotations) do
local effective = annotation.datetimeUpdated or annotation.datetime
if effective > ann_watermark then
table.insert(delta, annotation)
end
end
local pending = ctx.pending_books[entry.md5]
if pending then pending.ann_chunks_left = 0 end
for i = 1, #delta, ANN_BATCH_TOTAL do
local chunk_annotations = {}
for j = i, math.min(i + ANN_BATCH_TOTAL - 1, #delta) do
table.insert(chunk_annotations, delta[j])
end
ctx.ann_chunks:push({ md5 = entry.md5, annotations = chunk_annotations })
if pending then pending.ann_chunks_left = pending.ann_chunks_left + 1 end
end
end
ctx.ann_queue:clear()
end
local function stepAnnotationsLegacyNext(ctx)
if ctx.ann_chunks:isEmpty() then
step(ctx, ctx.steps.statesNext)
return
end
local books = {}
local total = 0
while not ctx.ann_chunks:isEmpty() and #books < ANN_BATCH_BOOKS do
local next_chunk = ctx.ann_chunks:peek()
if total + #next_chunk.annotations > ANN_BATCH_TOTAL then break end
ctx.ann_chunks:pop()
table.insert(books, { hash = next_chunk.md5, annotations = next_chunk.annotations })
total = total + #next_chunk.annotations
end
local body, err = ctx.client:uploadAnnotations(books)
if aborted(ctx) then return end
if not body then
if isAuthError(err) then return finish(ctx, "auth") end
logger.dbg("BookOrbit: annotations upload failed:", err)
ctx.had_errors = true
ctx.highlight_summary = BookOrbitHighlightSummary.add(ctx.highlight_summary, { had_errors = true })
for _, book in ipairs(books) do
local pending = ctx.pending_books[book.hash]
if pending then pending.failed = true end
end
step(ctx, ctx.steps.annotationsLegacyNext)
return
end
local unmatched = {}
for _, hash in ipairs(body.unmatched or {}) do
unmatched[hash] = true
end
for _, book in ipairs(books) do
local pending = ctx.pending_books[book.hash]
if unmatched[book.hash] then
ctx.state:setUnmatched(book.hash)
if pending then pending.failed = true end
elseif pending then
pending.ann_chunks_left = pending.ann_chunks_left - 1
ctx.counts.annotations = ctx.counts.annotations + #book.annotations
ctx.highlight_summary = BookOrbitHighlightSummary.add(ctx.highlight_summary, {
uploaded = #book.annotations,
})
end
end
step(ctx, ctx.steps.annotationsLegacyNext)
end
-- Phase 6: book status + rating, batched.
local function stepStatesNext(ctx)
if ctx.states_items:isEmpty() then
step(ctx, ctx.steps.progressNext)
return
end
local batch = {}
while #batch < STATES_BATCH do
local item = ctx.states_items:pop()
if not item then break end
table.insert(batch, item)
end
local body, err = ctx.client:uploadBookStates(batch)
if aborted(ctx) then return end
if not body then
if isAuthError(err) then return finish(ctx, "auth") end
logger.dbg("BookOrbit: book states upload failed:", err)
ctx.had_errors = true
for _, item in ipairs(batch) do
local pending = ctx.pending_books[item.hash]
if pending then pending.failed = true end
end
step(ctx, ctx.steps.statesNext)
return
end
local unmatched = {}
for _, hash in ipairs(body.unmatched or {}) do
unmatched[hash] = true
end
local results = {}
for _, result in ipairs(body.results or {}) do
results[result.hash] = result
end
for _, item in ipairs(batch) do
local pending = ctx.pending_books[item.hash]
if unmatched[item.hash] then
ctx.state:setUnmatched(item.hash)
if pending then pending.failed = true end
elseif pending then
-- A server-kept tie still counts as synced: the device value was considered.
pending.state_acked = true
local book = ctx.state:getBook(item.hash)
local server_state = BookOrbitSidecar.stateFromServerResult(results[item.hash])
if book and server_state then
if not server_state.status_known then
local payload = pending.state_payload or {}
BookOrbitSidecar.rememberUploadedState(book, {}, { status = payload.status, statusModified = payload.statusModified })
end
if book.file and book.file ~= currentlyOpenFile() then
local touched = BookOrbitSidecar.applyServerStateSidecar(book.file, server_state)
if touched then
pending.mtime = BookOrbitSidecar.sidecarMtime(book.file) or pending.mtime
end
elseif book.file and book.file == currentlyOpenFile() then
local ui = require("apps/reader/readerui").instance
BookOrbitSidecar.applyServerStateLive(ui, server_state)
end
BookOrbitSidecar.rememberServerState(book, server_state)
pending.state_used_server = true
end
end
end
step(ctx, ctx.steps.statesNext)
end
-- Phase 7: bulk progress for matched books whose sidecar percent changed.
local function stepProgressNext(ctx)
if ctx.progress_items:isEmpty() then
step(ctx, ctx.steps.done)
return
end
local batch = {}
while #batch < PROGRESS_BATCH do
local item = ctx.progress_items:pop()
if not item then break end
table.insert(batch, item)
end
local body, err = ctx.client:bulkProgress(batch)
if aborted(ctx) then return end
if not body then
if isAuthError(err) then return finish(ctx, "auth") end
logger.dbg("BookOrbit: bulk progress upload failed:", err)
ctx.had_errors = true
for _, item in ipairs(batch) do
local pending = ctx.pending_books[item.hash]
if pending then pending.failed = true end
end
step(ctx, ctx.steps.progressNext)
return
end
local unmatched = {}
for _, hash in ipairs(body.unmatched or {}) do
unmatched[hash] = true
end
for _, item in ipairs(batch) do
local pending = ctx.pending_books[item.hash]
if unmatched[item.hash] then
ctx.state:setUnmatched(item.hash)
if pending then pending.failed = true end
elseif pending then
pending.progress_acked = true
end
end
step(ctx, ctx.steps.progressNext)
end
-- Phase 8: commit per-book sidecar watermarks for fully acked books, record
-- the sweep server-side and store the fresh library version token.
local function stepDone(ctx)
local device_now = os.date("%Y-%m-%d %H:%M:%S")
for md5, pending in pairs(ctx.pending_books) do
local book = ctx.state:getBook(md5)
if book and not pending.failed then
local ann_done = pending.ann_done or (pending.ann_chunks_left ~= nil and pending.ann_chunks_left <= 0)
local state_done = not pending.need_state or pending.state_acked
local progress_done = not pending.need_progress or pending.progress_acked
if ann_done and state_done and progress_done then
book.sidecarMtime = pending.mtime
book.annCount = pending.ann_count
-- A copy the server cannot place highlights in sent none, so its
-- watermark stays behind them for a later compatible match.
if not book.positionIncompatible then
BookOrbitAnnotations.advanceWatermark(book, pending.ann_max_datetime, device_now)
end
if state_done and pending.need_state then
if not pending.state_used_server then
book.statusSyncedModified = pending.status_modified or book.statusSyncedModified
end
-- The sweep always sends the forced-pull payload, so an
-- acknowledged book has current server state and the
-- lifecycle path can stay quiet until the bound expires.
BookOrbitSidecar.markStatePulled(book)
if not pending.state_used_server then
BookOrbitSidecar.rememberUploadedState(book, pending.state_summary or {}, pending.state_payload)
end
end
if progress_done and pending.need_progress then
book.progressPushedPct = pending.percentage
end
end
end
end
local body = ctx.client:sweepComplete(ctx.counts)
if aborted(ctx) then return end
if body and body.libraryVersion then
ctx.server_library_version = body.libraryVersion
end
if ctx.server_library_version then
local known = ctx.state.global.libraryVersion
if ctx.full_recheck or known == nil then
ctx.state.global.needsFullRecheck = false
elseif ctx.server_library_version ~= known then
-- The library changed; the next sweep rechecks local hash mappings.
ctx.state.global.needsFullRecheck = true
end
ctx.state.global.libraryVersion = ctx.server_library_version
ctx.state.global.libraryVersionCheckedAt = os.time()
end
ctx.state.global.lastSweepAt = os.time()
finish(ctx)
end
function BookOrbitSweep.run(opts)
if BookOrbitSweep.running then
if opts.interactive then
require("bookorbit_toast").show(_("A sync is already running"), { icon = "hourglass" })
end
return false
end
local BookOrbitBookSync = require("bookorbit_book_sync")
if BookOrbitBookSync.isRunning() then
if opts.interactive then
require("bookorbit_toast").show(_("Syncing this book first. Try again shortly."), { icon = "hourglass" })
end
return false
end
local client = BookOrbitApi.new(opts.api)
if not client:isConfigured() then
if opts.interactive then
require("bookorbit_toast").error(_("Sign in to BookOrbit first."))
end
return false
end
BookOrbitSweep.running = true
-- Flush the open book (sidecar + statistics DB) so a mid-session sweep
-- sees current annotations, summary and reading time.
local rdr = require("apps/reader/readerui").instance
if rdr then
pcall(rdr.saveSettings, rdr)
end
local state = BookOrbitStateManager.session()
run_generation = run_generation + 1
local ctx = {
client = client,
state = state,
generation = run_generation,
interactive = opts.interactive or false,
quiet = opts.quiet == true,
on_progress = opts.on_progress,
annotation_sync = opts.annotation_sync ~= false,
plugin = opts.plugin,
-- An interactive run is no longer a reason on its own: "Sync all books"
-- is incremental, and the full rematch is its own explicit action.
full_recheck = opts.full_recheck == true or state.global.needsFullRecheck or false,
on_finish = opts.on_finish,
counts = { books_matched = 0, page_stats = 0, annotations = 0, ann_applied = 0, ann_deleted = 0 },
timing = {
enumerate_ms = 0,
sidecar_ms = 0,
partial_md5_ms = 0,
partial_md5_count = 0,
max_item_ms = 0,
},
highlight_summary = BookOrbitHighlightSummary.normalize{
event = "sweep",
reason = opts.interactive and "manual" or "auto",
},
had_errors = false,
}
ctx.stats_session = BookOrbitStatsReader.openSession()
ctx.steps = {
enumerateStats = stepEnumerateStats,
enumerateCandidates = stepEnumerateCandidates,
enumerateHistory = stepEnumerateHistory,
match = stepMatch,
matchNext = stepMatchNext,
statsNext = stepStatsNext,
sidecars = stepSidecars,
annotationsNext = stepAnnotationsNext,
annotationsLegacyNext = stepAnnotationsLegacyNext,
statesNext = stepStatesNext,
progressNext = stepProgressNext,
done = stepDone,
}
active_ctx = ctx
setProgress(ctx, ctx.full_recheck
and _("Rechecking every book against BookOrbit. This may take a while.")
or _("Syncing to BookOrbit. This may take a while on first run."), true,
ctx.full_recheck and _("Rechecking") or _("Starting"))
logger.info("BookOrbit: sweep started, interactive:", ctx.interactive,
"full recheck:", ctx.full_recheck)
step(ctx, ctx.steps.enumerateStats)
return true
end
return BookOrbitSweep