LinkPort/internal/syncjob/engine.go

// Package syncjob synchronizes configuration between devices.
package syncjob

import (
	"context"
	"errors"
	"linkport/internal/backup"
	"linkport/internal/store"
	"sync"
	"time"
)

var (
	ErrBusy     = errors.New("本机已有同步任务正在运行")
	ErrSettings = errors.New("请先保存有效的 WebDAV 同步设置")
)

type Result struct {
	Endpoint string `json:"endpoint"`
	Key      string `json:"key"`
	Action   string `json:"action"`
}

type Engine struct {
	store    Repository
	objects  func(backup.WebDAVConfig) backup.ObjectStore
	guard    func() func()
	canApply func() bool
	mu       sync.Mutex
	ctx      context.Context
	cancel   context.CancelFunc
	done     chan struct{}
	start    sync.Once
}

func New(s Repository, guard func() func(), canApply func() bool) *Engine {
	ctx, cancel := context.WithCancel(context.Background())
	return &Engine{store: s, objects: backup.NewWebDAVObjectStore, guard: guard, canApply: canApply, ctx: ctx, cancel: cancel, done: make(chan struct{})}
}

func (e *Engine) Start() {
	e.start.Do(func() {
		go func() {
			defer close(e.done)
			e.Check(e.ctx, time.Now())
			ticker := time.NewTicker(15 * time.Second)
			defer ticker.Stop()
			for {
				select {
				case <-e.ctx.Done():
					return
				case now := <-ticker.C:
					e.Check(e.ctx, now)
				}
			}
		}()
	})
}

func (e *Engine) Close() {
	e.cancel()
	e.Start()
	<-e.done
	e.mu.Lock()
	e.mu.Unlock()
}

func NextRun(c store.WebDAVSyncSettings, state store.WebDAVSyncState) time.Time {
	if !c.Automatic {
		return time.Time{}
	}
	base := c.UpdatedAt
	if last, err := time.Parse(time.RFC3339Nano, state.LastAttempt); err == nil && last.After(base) {
		base = last
	}
	return base.Add(time.Duration(c.IntervalMinutes) * time.Minute)
}

// A restart makes at most one overdue backup; missed intervals are not replayed.
func (e *Engine) Check(ctx context.Context, now time.Time) {
	_, _ = e.run(ctx, "scheduler", Request{}, true, now)
}

func (e *Engine) Run(ctx context.Context, source string, request Request) (Result, error) {
	return e.run(ctx, source, request, false, time.Now())
}

var ErrConflict = errors.New("两端配置均有变化，请选择保留本机或远端配置后重新同步")
var ErrRunning = errors.New("远端配置等待应用，请先停止本机全部网关后再次同步")

type Request struct {
	BackupOnly bool   `json:"-"`
	Resolution string `json:"resolution"`
	LocalHash  string `json:"localHash"`
	RemoteHash string `json:"remoteHash"`
}

func (e *Engine) run(parent context.Context, source string, request Request, scheduled bool, now time.Time) (result Result, err error) {
	if !e.mu.TryLock() {
		return result, ErrBusy
	}
	defer e.mu.Unlock()
	if e.ctx.Err() != nil {
		return result, errors.New("同步服务正在停止")
	}
	if e.guard != nil {
		unlock := e.guard()
		defer unlock()
	}
	c, err := e.store.WebDAVSyncSettings()
	if err != nil {
		return result, ErrSettings
	}
	state, err := e.store.WebDAVSyncState()
	if err != nil {
		return result, errors.New("无法读取同步状态")
	}
	if scheduled && (!c.Automatic || NextRun(c, state).After(now)) {
		return result, nil
	}
	if c.Validate() != nil {
		return result, ErrSettings
	}
	if request.Resolution != "" && request.Resolution != "local" && request.Resolution != "remote" {
		return result, errors.New("无效的同步冲突选择")
	}
	lease, ok, err := e.store.ClaimWebDAVSync(now, state.LastAttempt)
	if err != nil {
		return result, err
	}
	if !ok {
		return result, ErrBusy
	}
	defer func() {
		message := ""
		if err != nil {
			message = err.Error()
		}
		if finish := e.store.FinishWebDAVSync(lease, result.Endpoint, result.Key, result.Action, message); finish != nil && err == nil {
			err = errors.New("同步已执行，但本机状态保存失败，请重新同步")
		}
		if err != nil {
			e.store.Event("", source, "webdav_sync_failed", "")
		} else if result.Action != "unchanged" {
			e.store.Event("", source, "webdav_sync_"+result.Action, "")
		}
	}()
	ctx, cancel := context.WithTimeout(parent, 60*time.Second)
	stop := context.AfterFunc(e.ctx, cancel)
	defer func() { stop(); cancel() }()
	return e.transfer(ctx, c, state, request, now)
}