imap-copier/internal/orchestrator/orchestrator.go

package orchestrator

import (
	"context"
	"errors"
	"log/slog"
	"sync"
	"sync/atomic"
	"time"

	"github.com/vasyansk/imap-copier/internal/crypto"
	"github.com/vasyansk/imap-copier/internal/imapx"
	"github.com/vasyansk/imap-copier/internal/store"
	"github.com/vasyansk/imap-copier/internal/wshub"
)

var ErrNotTested = errors.New("accounts not fully tested")
var ErrAlreadyRunning = errors.New("task already running")
var ErrNoAccountsSelected = errors.New("no matching accounts selected")
var ErrNothingToResume = errors.New("no paused accounts to resume")

// maxAccountErrors caps how many individual error rows one account records per
// run, so a corrupt mailbox producing thousands of failures can't bloat the
// account_errors table. The cap'th row is a synthetic "further errors
// suppressed" note.
const maxAccountErrors = 500

// A running account that emits no scan/copy progress for stallTimeout is wedged
// (silent server mid-FETCH, stalled APPEND). The watchdog cancels it so the
// connections close and the worker unwinds instead of hanging forever. This
// replaces socket-level read deadlines, which can't tell an idle connection
// from a stuck one. The threshold is generous so slow-but-live runs aren't cut.
const (
	stallTimeout       = 3 * time.Minute
	stallCheckInterval = 30 * time.Second
)

// folderPlan is one source folder scheduled for copy and its destination name.
type folderPlan struct {
	src, dst string
	total    int64
}

// planFolders decides which of an account's source folders to copy and where.
// Excluded source folders are dropped; a folder absent from mapping keeps its
// own name. Order follows the input folder list.
func planFolders(folders []string, mapping map[string]string, excluded []string) []folderPlan {
	skip := make(map[string]struct{}, len(excluded))
	for _, e := range excluded {
		skip[e] = struct{}{}
	}
	plan := make([]folderPlan, 0, len(folders))
	for _, f := range folders {
		if _, ok := skip[f]; ok {
			continue
		}
		df := f
		if m, ok := mapping[f]; ok && m != "" {
			df = m
		}
		plan = append(plan, folderPlan{src: f, dst: df})
	}
	return plan
}

// A run stops either on its own or because the operator intervened. Pausing and
// cancelling take the same path — stop the in-flight work — and differ only in
// the status left behind: paused accounts are what Resume picks up again.
type stopReason int32

const (
	stopNone stopReason = iota
	stopPaused
	stopCancelled
)

func (r stopReason) accountStatus() string {
	if r == stopPaused {
		return "paused"
	}
	return "cancelled"
}

// runHandle is the live state of one task's run: the cancel that stops every
// account under it, plus why it was stopped.
type runHandle struct {
	cancel context.CancelFunc
	reason atomic.Int32
}

func (h *runHandle) stopWith(r stopReason) {
	h.reason.CompareAndSwap(int32(stopNone), int32(r))
	h.cancel()
}

func (h *runHandle) stopReason() stopReason { return stopReason(h.reason.Load()) }

type Orchestrator struct {
	store       *store.Store
	hub         *wshub.Hub
	encKey      []byte
	concurrency int

	mu      sync.Mutex
	cancels map[int64]context.CancelFunc // account_id -> cancel of its in-flight copy
	runs    map[int64]*runHandle         // task_id -> live run
}

func New(s *store.Store, hub *wshub.Hub, encKey []byte, concurrency int) *Orchestrator {
	return &Orchestrator{
		store: s, hub: hub, encKey: encKey, concurrency: concurrency,
		cancels: map[int64]context.CancelFunc{},
		runs:    map[int64]*runHandle{},
	}
}

// PauseTask stops the task's live run, leaving every unfinished account
// "paused" so ResumeTask can pick them up. Returns false if nothing is running.
func (o *Orchestrator) PauseTask(taskID int64) bool { return o.stopRun(taskID, stopPaused) }

// CancelTask stops the task's live run and marks every unfinished account
// "cancelled". Returns false if nothing is running.
func (o *Orchestrator) CancelTask(taskID int64) bool { return o.stopRun(taskID, stopCancelled) }

func (o *Orchestrator) stopRun(taskID int64, reason stopReason) bool {
	o.mu.Lock()
	h, ok := o.runs[taskID]
	o.mu.Unlock()
	if !ok {
		return false
	}
	h.stopWith(reason)
	return true
}

func (o *Orchestrator) registerRun(taskID int64, h *runHandle) {
	o.mu.Lock()
	o.runs[taskID] = h
	o.mu.Unlock()
}

func (o *Orchestrator) unregisterRun(taskID int64) {
	o.mu.Lock()
	delete(o.runs, taskID)
	o.mu.Unlock()
}

// ResumeTask restarts a paused task with exactly the accounts the pause left
// unfinished. Everything already copied is skipped by the migration journal, so
// each account continues where it stopped.
func (o *Orchestrator) ResumeTask(ctx context.Context, taskID int64) (int64, error) {
	accs, err := o.store.ListAccountsByTask(ctx, taskID)
	if err != nil {
		return 0, err
	}
	ids := make([]int64, 0, len(accs))
	for _, a := range accs {
		if a.Status == "paused" {
			ids = append(ids, a.ID)
		}
	}
	if len(ids) == 0 {
		return 0, ErrNothingToResume
	}
	return o.Run(ctx, taskID, "manual", ids)
}

// CancelAccount aborts the in-flight copy for one account, if it is running.
// Returns true if a running copy was found and signalled to stop.
func (o *Orchestrator) CancelAccount(accountID int64) bool {
	o.mu.Lock()
	cancel, ok := o.cancels[accountID]
	o.mu.Unlock()
	if ok {
		cancel()
	}
	return ok
}

func (o *Orchestrator) registerCancel(accountID int64, cancel context.CancelFunc) {
	o.mu.Lock()
	o.cancels[accountID] = cancel
	o.mu.Unlock()
}

func (o *Orchestrator) unregisterCancel(accountID int64) {
	o.mu.Lock()
	delete(o.cancels, accountID)
	o.mu.Unlock()
}

// selectAccounts narrows accs to those whose ID is in ids, preserving input
// order. An empty or nil ids means "all accounts" — the scheduler and the
// unfiltered manual run rely on this.
func selectAccounts(accs []store.Account, ids []int64) []store.Account {
	if len(ids) == 0 {
		return accs
	}
	want := make(map[int64]struct{}, len(ids))
	for _, id := range ids {
		want[id] = struct{}{}
	}
	out := make([]store.Account, 0, len(ids))
	for _, a := range accs {
		if _, ok := want[a.ID]; ok {
			out = append(out, a)
		}
	}
	return out
}

func gateOK(accs []store.Account) bool {
	if len(accs) == 0 {
		return false
	}
	for _, a := range accs {
		if a.TestSrcStatus != "ok" || a.TestDstStatus != "ok" {
			return false
		}
	}
	return true
}

func (o *Orchestrator) endpoints(ctx context.Context, task store.Task) (imapx.Endpoint, imapx.Endpoint, error) {
	src, err := o.store.GetEndpoint(ctx, task.SrcEndpointID)
	if err != nil {
		return imapx.Endpoint{}, imapx.Endpoint{}, err
	}
	dst, err := o.store.GetEndpoint(ctx, task.DstEndpointID)
	if err != nil {
		return imapx.Endpoint{}, imapx.Endpoint{}, err
	}
	toEP := func(e store.Endpoint) imapx.Endpoint {
		return imapx.Endpoint{Host: e.Host, Port: e.Port, TLSMode: e.TLSMode}
	}
	return toEP(src), toEP(dst), nil
}

func (o *Orchestrator) TestAccounts(ctx context.Context, taskID int64) error {
	task, err := o.store.GetTask(ctx, taskID)
	if err != nil {
		return err
	}
	srcEP, dstEP, err := o.endpoints(ctx, task)
	if err != nil {
		return err
	}
	accs, err := o.store.ListAccountsByTask(ctx, taskID)
	if err != nil {
		return err
	}
	for _, a := range accs {
		o.testSide(ctx, srcEP, a.ID, "src", a.SrcLogin, a.SrcPassEnc, taskID)
		o.testSide(ctx, dstEP, a.ID, "dst", a.DstLogin, a.DstPassEnc, taskID)
	}
	return nil
}

func (o *Orchestrator) testSide(ctx context.Context, ep imapx.Endpoint, accID int64, side, login, passEnc string, taskID int64) {
	status := "ok"
	errMsg := ""
	pass, err := crypto.Decrypt(o.encKey, passEnc)
	if err == nil {
		_, err = imapx.TestLogin(ctx, ep, login, string(pass))
	}
	if err != nil {
		status = "fail"
		errMsg = err.Error()
		slog.Warn("account test failed", "account", accID, "side", side,
			"login", login, "host", ep.Host, "port", ep.Port, "err", err)
	}
	_ = o.store.SetAccountTestStatus(ctx, accID, side, status)
	o.hub.Publish(wshub.Event{Type: "account_test", TaskID: taskID,
		Data: map[string]any{
			"account_id": accID, "side": side, "status": status,
			"login": login, "host": ep.Host, "port": ep.Port, "error": errMsg,
		}})
}

// shouldBreak reports whether a completed run should trip the schedule breaker:
// only scheduled runs that ended with errors.
func shouldBreak(trigger string, totErr int64) bool {
	return trigger == "scheduled" && totErr > 0
}

func (o *Orchestrator) Run(ctx context.Context, taskID int64, trigger string, accountIDs []int64) (int64, error) {
	task, err := o.store.GetTask(ctx, taskID)
	if err != nil {
		return 0, err
	}
	accs, err := o.store.ListAccountsByTask(ctx, taskID)
	if err != nil {
		return 0, err
	}
	accs = selectAccounts(accs, accountIDs)
	// A non-empty request that matched nothing is a client error, distinct
	// from "not tested".
	if len(accountIDs) > 0 && len(accs) == 0 {
		return 0, ErrNoAccountsSelected
	}
	if !gateOK(accs) {
		return 0, ErrNotTested
	}
	acquired, err := o.store.TryMarkTaskRunning(ctx, taskID)
	if err != nil {
		return 0, err
	}
	if !acquired {
		return 0, ErrAlreadyRunning
	}
	srcEP, dstEP, err := o.endpoints(ctx, task)
	if err != nil {
		_ = o.store.SetTaskStatus(ctx, taskID, "error")
		return 0, err
	}
	runID, err := o.store.CreateRun(ctx, taskID, trigger)
	if err != nil {
		_ = o.store.SetTaskStatus(ctx, taskID, "error")
		return 0, err
	}
	o.hub.Publish(wshub.Event{Type: "run_started", TaskID: taskID, Data: map[string]any{"run_id": runID}})

	// dbCtx outlives the request so status/counter writes still land after a
	// pause or cancel; runCtx is what Pause/Cancel actually stop, and every
	// account's IMAP work hangs off it.
	dbCtx := context.WithoutCancel(ctx)
	runCtx, runCancel := context.WithCancel(dbCtx)
	h := &runHandle{cancel: runCancel}
	o.registerRun(taskID, h)

	go o.runAll(dbCtx, runCtx, h, task, runID, accs, srcEP, dstEP, trigger)
	return runID, nil
}

func (o *Orchestrator) runAll(ctx, runCtx context.Context, h *runHandle, task store.Task, runID int64, accs []store.Account, srcEP, dstEP imapx.Endpoint, trigger string) {
	defer o.unregisterRun(task.ID)
	defer h.cancel()
	defer func() {
		if r := recover(); r != nil {
			slog.Error("run coordinator panicked", "task", task.ID, "run", runID, "panic", r)
			_ = o.store.FinishRun(ctx, runID, "error", 0, 0, 0)
			_ = o.store.SetTaskStatus(ctx, task.ID, "error")
			if trigger == "scheduled" {
				_ = o.store.SetTaskBroken(ctx, task.ID)
				o.hub.Publish(wshub.Event{Type: "task_broken", TaskID: task.ID,
					Data: map[string]any{"task_id": task.ID, "errors": int64(0)}})
			}
		}
	}()

	var (
		mu                            sync.Mutex
		totCopied, totSkipped, totErr int64
	)
	sem := make(chan struct{}, o.concurrency)
	var wg sync.WaitGroup

	for i, a := range accs {
		// Stopped mid-queue: the accounts that never started are marked with the
		// same status as the ones that were interrupted, so a pause leaves the
		// whole remainder resumable and a cancel leaves it cancelled.
		if runCtx.Err() != nil {
			st := h.stopReason().accountStatus()
			for _, rest := range accs[i:] {
				_ = o.store.SetAccountStatus(ctx, rest.ID, st)
			}
			break
		}
		wg.Add(1)
		sem <- struct{}{}
		go func(a store.Account) {
			defer wg.Done()
			defer func() { <-sem }()
			defer func() {
				if r := recover(); r != nil {
					slog.Error("account worker panicked", "task", task.ID, "account", a.ID, "panic", r)
					_ = o.store.SetAccountStatus(ctx, a.ID, "error")
					o.hub.Publish(wshub.Event{Type: "error", TaskID: task.ID,
						Data: map[string]any{"account_id": a.ID, "error": "internal panic"}})
					mu.Lock()
					totErr++
					mu.Unlock()
				}
			}()
			c, s, e := o.runAccount(ctx, runCtx, h, task, runID, a, srcEP, dstEP)
			mu.Lock()
			totCopied += c
			totSkipped += s
			totErr += e
			mu.Unlock()
		}(a)
	}
	wg.Wait()

	reason := h.stopReason()
	status := "done"
	switch {
	case reason == stopPaused:
		status = "paused"
	case reason == stopCancelled:
		status = "cancelled"
	case totErr > 0:
		status = "done_with_errors"
	}
	_ = o.store.FinishRun(ctx, runID, status, totCopied, totSkipped, totErr)
	_ = o.store.SetTaskStatus(ctx, task.ID, status)
	o.hub.Publish(wshub.Event{Type: "run_done", TaskID: task.ID,
		Data: map[string]any{"run_id": runID, "status": status,
			"copied": totCopied, "skipped": totSkipped, "errors": totErr}})

	// An operator stopping the run is not a schedule failure, so leave the
	// breaker alone even when the accounts that did run reported errors.
	if reason == stopNone && shouldBreak(trigger, totErr) {
		_ = o.store.SetTaskBroken(ctx, task.ID)
		o.hub.Publish(wshub.Event{Type: "task_broken", TaskID: task.ID,
			Data: map[string]any{"task_id": task.ID, "errors": totErr}})
	}
}

func (o *Orchestrator) runAccount(ctx, runCtx context.Context, h *runHandle, task store.Task, runID int64, a store.Account, srcEP, dstEP imapx.Endpoint) (int64, int64, int64) {
	o.hub.Publish(wshub.Event{Type: "account_started", TaskID: task.ID, Data: map[string]any{
		"account_id": a.ID,
		"src_login":  a.SrcLogin, "src_host": srcEP.Host, "src_port": srcEP.Port,
		"dst_login": a.DstLogin, "dst_host": dstEP.Host, "dst_port": dstEP.Port,
	}})
	_ = o.store.SetAccountStatus(ctx, a.ID, "running")
	_ = o.store.SetAccountError(ctx, a.ID, "")  // clear any error from a previous run
	_ = o.store.ResetAccountCounters(ctx, a.ID) // start from zero; IncAccountCounters is additive
	_ = o.store.ClearAccountErrors(ctx, a.ID)   // drop last run's per-error rows

	// Per-account cancellable context: IMAP work uses actx, so both CancelAccount
	// and a task-wide pause/cancel (which cancels runCtx) stop it. DB writes keep
	// ctx — the uncancellable one from runAll — so status/counters persist even
	// after cancellation.
	actx, cancel := context.WithCancel(runCtx)
	o.registerCancel(a.ID, cancel)
	defer func() {
		o.unregisterCancel(a.ID)
		cancel()
	}()

	srcPass, err := crypto.Decrypt(o.encKey, a.SrcPassEnc)
	if err != nil {
		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "src", err)
	}
	dstPass, err := crypto.Decrypt(o.encKey, a.DstPassEnc)
	if err != nil {
		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "dst", err)
	}

	src, err := imapx.Connect(actx, srcEP)
	if err != nil {
		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "src", err)
	}
	if err := src.Login(a.SrcLogin, string(srcPass)).Wait(); err != nil {
		_ = src.Logout().Wait()
		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "src", err)
	}
	// srcClient holds the LIVE source connection. A body-read timeout (server
	// under-delivering a literal) forces a mid-run reconnect via reconnectSrc,
	// which swaps this pointer. The cancel goroutine and the deferred logout
	// below both read through it, so they always act on the current connection
	// rather than a stale one that was already replaced and logged out.
	var srcClient atomic.Pointer[imapx.Client]
	srcClient.Store(src)
	defer func() { _ = srcClient.Load().Logout().Wait() }()

	dst, err := imapx.Connect(actx, dstEP)
	if err != nil {
		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "dst", err)
	}
	defer func() { _ = dst.Logout().Wait() }()
	if err := dst.Login(a.DstLogin, string(dstPass)).Wait(); err != nil {
		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "dst", err)
	}

	// reconnectSrc dials and logs in a fresh source client, swaps it in as the
	// live connection, and logs the old (desynced) one out. CopyFolder calls it
	// to recover after a message body times out: the server left the connection
	// mid-literal, so it can't be reused. Bound to actx, so a cancelled account
	// fails the dial instead of reconnecting.
	reconnectSrc := func() (*imapx.Client, error) {
		nc, err := imapx.Connect(actx, srcEP)
		if err != nil {
			return nil, err
		}
		if err := nc.Login(a.SrcLogin, string(srcPass)).Wait(); err != nil {
			_ = nc.Logout().Wait()
			return nil, err
		}
		if old := srcClient.Swap(nc); old != nil {
			_ = old.Logout().Wait()
		}
		slog.Info("reconnected src after message body timeout", "account", a.ID, "src_login", a.SrcLogin)
		return nc, nil
	}

	// On cancel, close the connections so any in-flight network read (a slow
	// FETCH/Collect that ctx.Err() checks can't interrupt) unblocks immediately.
	go func() {
		<-actx.Done()
		_ = srcClient.Load().Close()
		_ = dst.Close()
	}()

	// Keep the destination connection warm. It sits idle for the whole
	// source-side metadata scan (Pass 1 of CopyFolder), which on a large
	// mailbox runs for minutes; without traffic the server drops it and every
	// subsequent APPEND fails with "use of closed network connection", copying
	// nothing. Periodic NOOPs prevent that. Only dst needs it — src is
	// continuously busy scanning/fetching. Bound to actx so it stops with the
	// account.
	go imapx.Keepalive(actx, dst, imapx.KeepaliveInterval)

	// Progress watchdog: track the last time we saw scan/copy activity; if it
	// goes quiet for stallTimeout, cancel the account so the connections close
	// and this worker unwinds (it would otherwise block forever on a silent
	// server). touch() is called on every progress signal below.
	var lastActivity atomic.Int64
	lastActivity.Store(time.Now().UnixNano())
	touch := func() { lastActivity.Store(time.Now().UnixNano()) }
	go func() {
		t := time.NewTicker(stallCheckInterval)
		defer t.Stop()
		for {
			select {
			case <-actx.Done():
				return
			case <-t.C:
				if time.Since(time.Unix(0, lastActivity.Load())) > stallTimeout {
					slog.Warn("account stalled with no progress; cancelling",
						"account", a.ID, "src_login", a.SrcLogin, "stall", stallTimeout)
					_ = o.store.SetAccountError(ctx, a.ID, "stalled: no progress for "+stallTimeout.String()+", cancelled")
					cancel()
					return
				}
			}
		}
	}()

	folders, err := imapx.ListFolders(src)
	touch()
	if err != nil {
		return o.accountFailed(ctx, runCtx, h, task.ID, runID, a, srcEP, dstEP, "src", err)
	}

	// Planning pass: decide folders from the account's own config, then EXAMINE
	// each to learn message counts for the overall progress bar.
	plan := planFolders(folders, a.FolderMapping, a.ExcludedFolders)
	var grandTotal int64
	for i := range plan {
		if actx.Err() != nil {
			break
		}
		n, cerr := imapx.FolderMessageCount(src, plan[i].src)
		if cerr != nil && actx.Err() == nil {
			slog.Warn("count folder failed", "account", a.ID, "folder", plan[i].src, "err", cerr)
		}
		plan[i].total = n
		grandTotal += n
		touch()
	}
	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;
	// c/s inside OnProgress are cumulative within the current folder.
	var baseCopied, baseSkipped int64
	var curFolder string
	var curTotal int64
	var lastEmit, lastScanEmit time.Time
	// Persist individual errors for the per-account error modal, capped so a
	// corrupt mailbox can't write unbounded rows. Runs on this goroutine, so
	// the counter is race-free (same reasoning as the progress vars above).
	var persistedErrs int
	addErr := func(kind, folder, ref, msg string) {
		persistedErrs++
		switch {
		case persistedErrs < maxAccountErrors:
			_ = o.store.AddAccountError(ctx, a.ID, runID, kind, folder, ref, msg)
		case persistedErrs == maxAccountErrors:
			_ = o.store.AddAccountError(ctx, a.ID, runID, "account", "", "",
				"too many errors — further errors suppressed")
		}
	}
	deps := imapx.CopyDeps{
		IsMigrated:   func(k string) (bool, error) { return o.store.IsMigrated(ctx, a.ID, k) },
		MarkMigrated: func(folder, k string) error { return o.store.MarkMigrated(ctx, a.ID, folder, k) },
		OnError:      func(ref, msg string) { addErr("message", curFolder, ref, msg) },
		// Fires as bytes move within a single message's FETCH/APPEND, so the
		// stall watchdog sees a large-but-live transfer as progress instead of
		// cancelling it as a wedged connection.
		OnActivity: touch,
		// Recovers from a message body timeout (server under-delivering a
		// literal) by swapping in a fresh source connection so the folder can
		// resume with the remaining messages.
		ReconnectSrc: reconnectSrc,
		OnProgress: func(c, s int) {
			touch()
			now := time.Now()
			done := c + s
			// throttle to ~3/sec per account, but always emit folder completion
			if now.Sub(lastEmit) < 350*time.Millisecond && int64(done) < curTotal {
				return
			}
			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_total": grandTotal,
			}})
		},
		// Fires after EXAMINE (before the long fetch) with the folder's message count.
		OnFolder: func(srcFolder, dstFolder string, total int64) {
			touch()
			curFolder, curTotal = srcFolder, total
			o.hub.Publish(wshub.Event{Type: "folder", TaskID: task.ID, Data: map[string]any{
				"account_id": a.ID, "src_login": a.SrcLogin,
				"folder": srcFolder, "dst_folder": dstFolder, "messages": total,
			}})
		},
		// Fires while streaming metadata (dedup scan) so the UI shows movement
		// before bodies start copying. Throttled to ~4/sec, always emit the last.
		OnScan: func(scanned, total int64) {
			touch()
			now := time.Now()
			if now.Sub(lastScanEmit) < 250*time.Millisecond && scanned < total {
				return
			}
			lastScanEmit = now
			o.hub.Publish(wshub.Event{Type: "scan", TaskID: task.ID, Data: map[string]any{
				"account_id": a.ID, "folder": curFolder, "scanned": scanned, "folder_total": total,
			}})
		},
	}
	for _, fp := range plan {
		if actx.Err() != nil {
			break // cancelled — stop scheduling more folders
		}
		res, err := imapx.CopyFolder(actx, srcClient.Load(), dst, fp.src, fp.dst, deps)
		folderErr := int64(0)
		if err != nil && actx.Err() == nil {
			slog.Warn("folder copy error", "account", a.ID, "src_login", a.SrcLogin, "folder", fp.src, "err", err)
			folderErr = 1
			_ = o.store.SetAccountError(ctx, a.ID, "folder \""+fp.src+"\": "+err.Error())
			addErr("folder", fp.src, "", err.Error())
			o.hub.Publish(wshub.Event{Type: "error", TaskID: task.ID, Data: map[string]any{
				"account_id": a.ID, "src_login": a.SrcLogin, "folder": fp.src, "error": err.Error(),
			}})
		}
		copied += int64(res.Copied)
		skipped += int64(res.Skipped)
		errs += int64(res.Errors) + folderErr
		baseCopied += int64(res.Copied)
		baseSkipped += int64(res.Skipped)
		// Persist message-level AND folder-level errors so the account row's
		// error count matches the task status (done_with_errors).
		_ = o.store.IncAccountCounters(ctx, a.ID, int64(res.Copied), int64(res.Skipped), int64(res.Errors)+folderErr)
	}

	if actx.Err() != nil {
		// A task-wide pause leaves the account resumable; anything else (per-account
		// cancel, stall watchdog, task-wide cancel) leaves it cancelled.
		st := "cancelled"
		if runCtx.Err() != nil {
			st = h.stopReason().accountStatus()
		}
		_ = o.store.SetAccountStatus(ctx, a.ID, st)
		o.hub.Publish(wshub.Event{Type: st, TaskID: task.ID,
			Data: map[string]any{"account_id": a.ID, "src_login": a.SrcLogin,
				"copied": copied, "skipped": skipped, "errors": errs}})
		slog.Info("account stopped", "account", a.ID, "src_login", a.SrcLogin,
			"status", st, "copied", copied, "skipped", skipped)
		return copied, skipped, errs
	}

	acctStatus := "done"
	if errs > 0 {
		acctStatus = "done_with_errors"
	}
	_ = o.store.SetAccountStatus(ctx, a.ID, acctStatus)
	o.hub.Publish(wshub.Event{Type: "account_done", TaskID: task.ID,
		Data: map[string]any{"account_id": a.ID, "src_login": a.SrcLogin, "dst_login": a.DstLogin,
			"copied": copied, "skipped": skipped, "errors": errs}})
	slog.Info("account copied", "account", a.ID, "src_login", a.SrcLogin, "copied", copied, "skipped", skipped, "errors", errs)
	return copied, skipped, errs
}

func (o *Orchestrator) accountFailed(ctx, runCtx context.Context, h *runHandle, taskID, runID int64, a store.Account, srcEP, dstEP imapx.Endpoint, side string, err error) (int64, int64, int64) {
	// A cancellation surfacing as an error is a stop, not a failure — and a
	// task-wide pause must still leave the account resumable.
	if errors.Is(err, context.Canceled) {
		st := "cancelled"
		if runCtx.Err() != nil {
			st = h.stopReason().accountStatus()
		}
		_ = o.store.SetAccountStatus(ctx, a.ID, st)
		o.hub.Publish(wshub.Event{Type: st, TaskID: taskID,
			Data: map[string]any{"account_id": a.ID, "src_login": a.SrcLogin}})
		return 0, 0, 0
	}
	login, host, port := a.SrcLogin, srcEP.Host, srcEP.Port
	if side == "dst" {
		login, host, port = a.DstLogin, dstEP.Host, dstEP.Port
	}
	slog.Error("account failed", "account", a.ID, "side", side, "login", login, "host", host, "port", port, "err", err)
	_ = o.store.SetAccountStatus(ctx, a.ID, "error")
	failMsg := side + " " + login + "@" + host + ": " + err.Error()
	_ = o.store.SetAccountError(ctx, a.ID, failMsg)
	_ = o.store.AddAccountError(ctx, a.ID, runID, "account", "", "", failMsg)
	o.hub.Publish(wshub.Event{Type: "error", TaskID: taskID,
		Data: map[string]any{"account_id": a.ID, "side": side, "login": login, "host": host, "port": port, "error": err.Error()}})
	return 0, 0, 1
}