feat: planning pass counts all folders up front for account-wide progress
Before copying, EXAMINE every folder to sum the account's total message count and emit a 'plan' event; progress events now carry account_total so the UI shows a real overall bar, percent and ETA (not just per-folder). - imapx.FolderMessageCount: read-only count of a folder - orchestrator: plan pass + grandTotal, plan event, account_total in progress - web: live progress keyed on account total; PLAN log line; overall bar/ETA Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MMHQTtnQtQqL8muAXHr9kd
This commit is contained in:
@@ -267,6 +267,33 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in
|
||||
return o.accountFailed(ctx, task.ID, a, srcEP, dstEP, "src", err)
|
||||
}
|
||||
|
||||
// Planning pass: EXAMINE every folder up front to learn the total message
|
||||
// count, so the UI can show an accurate overall bar / ETA before copying.
|
||||
type folderPlan struct {
|
||||
src, dst string
|
||||
total int64
|
||||
}
|
||||
plan := make([]folderPlan, 0, len(folders))
|
||||
var grandTotal int64
|
||||
for _, folder := range folders {
|
||||
if actx.Err() != nil {
|
||||
break
|
||||
}
|
||||
df := folder
|
||||
if m, ok := task.FolderMapping[folder]; ok {
|
||||
df = m
|
||||
}
|
||||
n, cerr := imapx.FolderMessageCount(src, folder)
|
||||
if cerr != nil && actx.Err() == nil {
|
||||
slog.Warn("count folder failed", "account", a.ID, "folder", folder, "err", cerr)
|
||||
}
|
||||
plan = append(plan, folderPlan{src: folder, dst: df, total: n})
|
||||
grandTotal += n
|
||||
}
|
||||
o.hub.Publish(wshub.Event{Type: "plan", TaskID: task.ID, Data: map[string]any{
|
||||
"account_id": a.ID, "src_login": a.SrcLogin, "folders": len(plan), "total": grandTotal,
|
||||
}})
|
||||
|
||||
var copied, skipped, errs int64
|
||||
// Account-level live progress state (all callbacks run on this goroutine,
|
||||
// so plain vars are race-free). base* = totals from completed folders;
|
||||
@@ -287,12 +314,13 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in
|
||||
}
|
||||
lastEmit = now
|
||||
o.hub.Publish(wshub.Event{Type: "progress", TaskID: task.ID, Data: map[string]any{
|
||||
"account_id": a.ID,
|
||||
"copied": baseCopied + int64(c),
|
||||
"skipped": baseSkipped + int64(s),
|
||||
"folder": curFolder,
|
||||
"folder_done": done,
|
||||
"folder_total": curTotal,
|
||||
"account_id": a.ID,
|
||||
"copied": baseCopied + int64(c),
|
||||
"skipped": baseSkipped + int64(s),
|
||||
"folder": curFolder,
|
||||
"folder_done": done,
|
||||
"folder_total": curTotal,
|
||||
"account_total": grandTotal,
|
||||
}})
|
||||
},
|
||||
// Fires after EXAMINE (before the long fetch) with the folder's message count.
|
||||
@@ -304,20 +332,16 @@ func (o *Orchestrator) runAccount(ctx context.Context, task store.Task, runID in
|
||||
}})
|
||||
},
|
||||
}
|
||||
for _, folder := range folders {
|
||||
for _, fp := range plan {
|
||||
if actx.Err() != nil {
|
||||
break // cancelled — stop scheduling more folders
|
||||
}
|
||||
dstFolder := folder
|
||||
if m, ok := task.FolderMapping[folder]; ok {
|
||||
dstFolder = m
|
||||
}
|
||||
res, err := imapx.CopyFolder(actx, src, dst, folder, dstFolder, deps)
|
||||
res, err := imapx.CopyFolder(actx, src, dst, fp.src, fp.dst, deps)
|
||||
if err != nil && actx.Err() == nil {
|
||||
slog.Warn("folder copy error", "account", a.ID, "src_login", a.SrcLogin, "folder", folder, "err", err)
|
||||
slog.Warn("folder copy error", "account", a.ID, "src_login", a.SrcLogin, "folder", fp.src, "err", err)
|
||||
errs++
|
||||
o.hub.Publish(wshub.Event{Type: "error", TaskID: task.ID, Data: map[string]any{
|
||||
"account_id": a.ID, "src_login": a.SrcLogin, "folder": folder, "error": err.Error(),
|
||||
"account_id": a.ID, "src_login": a.SrcLogin, "folder": fp.src, "error": err.Error(),
|
||||
}})
|
||||
}
|
||||
copied += int64(res.Copied)
|
||||
|
||||
Reference in New Issue
Block a user