sync

package
v0.0.0-...-d7033ae Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Apr 24, 2026 License: MIT Imports: 20 Imported by: 0

README

CloudPull Sync Engine

Overview

The sync engine is the heart of CloudPull, orchestrating the synchronization of Google Drive folders to local storage.

Architecture

Components

  1. Engine (engine.go)

    • Main orchestrator for sync operations
    • Manages sync sessions and lifecycle
    • Coordinates walker and downloader

  2. Walker (walker.go)

    • Traverses Google Drive folder structures
    • Memory-efficient pagination
    • Supports BFS and DFS strategies

  3. Downloader (downloader.go)

    • Manages file downloads with resume support
    • Handles Google Docs export
    • Checksum verification

  4. Worker (worker.go)

    • Concurrent download workers
    • Priority queue processing
    • Health monitoring and recovery

  5. Progress (progress.go)

    • Real-time progress tracking
    • Event emission for UI updates
    • Bandwidth calculation

Usage Example

// Initialize dependencies
logger := logger.NewLogger(logConfig)
db, _ := state.NewDatabase(dbConfig)
stateManager := state.NewManager(db, logger)
apiClient := api.NewClient(authConfig, logger)
progressTracker := progress.NewTracker()

// Create sync engine
engine := sync.NewEngine(sync.Config{
    StateManager:    stateManager,
    APIClient:       apiClient,
    Logger:          logger,
    ProgressTracker: progressTracker,
    // ... other config
})

// Start sync
ctx := context.Background()
sessionID, err := engine.StartSync(ctx, "drive-folder-id", "/local/path")

// Or resume existing sync
err = engine.ResumeSync(ctx, sessionID)

Features

Memory Efficiency
  • Streams folder contents without loading the entire tree
  • Pagination prevents memory overflow
  • Batch processing for database operations
Reliability
  • Automatic retry with exponential backoff
  • Resume from exact byte offset
  • Checksum verification
  • Atomic file operations
Performance
  • Concurrent downloads (configurable)
  • Priority queue (smallest files first)
  • Bandwidth throttling
  • Progress batching
Monitoring
  • Real-time progress events
  • Detailed error logging
  • Performance metrics
  • Health checks

Configuration

type Config struct {
    MaxConcurrentDownloads int
    DownloadChunkSize      int64
    BandwidthLimit         int64
    RetryAttempts          int
    RetryBackoff           time.Duration
    TempDir                string
    VerifyChecksums        bool
}

Error Handling

The sync engine uses the centralized error handler for:

  • Network errors: Retry with backoff
  • API quota errors: Longer backoff
  • Permission errors: Skip file
  • Storage errors: Pause sync
  • Corruption: Re-download

Events

Subscribe to sync events:

progressTracker.Subscribe(func(snapshot progress.Snapshot) {
    fmt.Printf("Progress: %.2f%% (%.2f MB/s)\n",
        snapshot.Percentage,
        snapshot.BytesPerSecond/1024/1024)
})

Event types:

  • FileStarted
  • FileProgress
  • FileCompleted
  • FileFailed
  • FolderStarted
  • FolderCompleted
  • SyncPaused
  • SyncResumed

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type DownloadInfo 

type DownloadInfo struct {
	StartTime       time.Time
	FileID          string
	FileName        string
	TempPath        string
	FinalPath       string
	Checksum        string
	ExportFormat    string
	Size            int64
	BytesDownloaded int64
	IsGoogleDoc     bool
}

DownloadInfo tracks active download information.

type DownloadManager 

type DownloadManager struct {
	// contains filtered or unexported fields
}

DownloadManager manages file downloads with advanced features.

func NewDownloadManager 

func NewDownloadManager(
	client *api.DriveClient,
	stateManager *state.Manager,
	progressTracker *ProgressTracker,
	errorHandler *errors.Handler,
	logger *logger.Logger,
	config *DownloadManagerConfig,
) (*DownloadManager, error)

NewDownloadManager creates a new download manager.

func (*DownloadManager) DownloadFile 

func (dm *DownloadManager) DownloadFile(ctx context.Context, file *state.File) error

DownloadFile downloads a single file with resume support.

func (*DownloadManager) GetStats 

func (dm *DownloadManager) GetStats() *DownloadManagerStats

GetStats returns download manager statistics.

func (*DownloadManager) ScheduleBatch 

func (dm *DownloadManager) ScheduleBatch(files []*state.File) error

ScheduleBatch schedules a batch of files for download.

func (*DownloadManager) ScheduleDownload 

func (dm *DownloadManager) ScheduleDownload(file *state.File, priority int) error

ScheduleDownload schedules a file for download.

func (*DownloadManager) Start 

func (dm *DownloadManager) Start(ctx context.Context) error

Start starts the download manager.

func (*DownloadManager) Stop 

func (dm *DownloadManager) Stop() error

Stop stops the download manager.

type DownloadManagerConfig 

type DownloadManagerConfig struct {
	TempDir         string
	ChunkSize       int64
	MaxConcurrent   int
	VerifyChecksums bool
}

DownloadManagerConfig contains configuration for the download manager.

func DefaultDownloadManagerConfig 

func DefaultDownloadManagerConfig() *DownloadManagerConfig

DefaultDownloadManagerConfig returns default configuration.

type DownloadManagerStats 

type DownloadManagerStats struct {
	WorkerPoolStats    *WorkerPoolStats
	TotalDownloads     int64
	ActiveDownloads    int64
	CompletedDownloads int64
	FailedDownloads    int64
	BytesDownloaded    int64
	AverageSpeed       int64
	AverageDuration    time.Duration
}

DownloadManagerStats contains download manager statistics.

type DownloadStats 

type DownloadStats struct {
	TotalDownloads     int64
	ActiveDownloads    int64
	CompletedDownloads int64
	FailedDownloads    int64
	BytesDownloaded    int64
	TotalDuration      time.Duration
	// contains filtered or unexported fields
}

DownloadStats tracks download statistics.

func (*DownloadStats) AddBytes 

func (ds *DownloadStats) AddBytes(n int64)

AddBytes atomically adds n bytes to the total bytes downloaded.

func (*DownloadStats) CompleteDownload 

func (ds *DownloadStats) CompleteDownload(bytes int64, duration time.Duration)

CompleteDownload updates all completion-related fields together.

func (*DownloadStats) DecrementActiveDownloads 

func (ds *DownloadStats) DecrementActiveDownloads()

DecrementActiveDownloads decrements the active download count by 1.

func (*DownloadStats) IncrementDownloads 

func (ds *DownloadStats) IncrementDownloads()

IncrementDownloads atomically increments the total downloads count by 1.

func (*DownloadStats) IncrementDownloadsAndActive 

func (ds *DownloadStats) IncrementDownloadsAndActive()

IncrementDownloadsAndActive increments the total and active download counts together.

func (*DownloadStats) IncrementFailedDownloads 

func (ds *DownloadStats) IncrementFailedDownloads()

IncrementFailedDownloads increments the failed download count by 1.

type DownloadTask 

type DownloadTask struct {
	CreatedAt   time.Time
	LastError   error
	File        *state.File
	StartedAt   *time.Time
	CompletedAt *time.Time
	Priority    int
	Retries     int
}

DownloadTask represents a file download task.

type Engine 

type Engine struct {
	// contains filtered or unexported fields
}

Engine is the main sync orchestrator.

func NewEngine 

func NewEngine(
	client *api.DriveClient,
	stateManager *state.Manager,
	errorHandler *errors.Handler,
	logger *logger.Logger,
	config *EngineConfig,
) (*Engine, error)

NewEngine creates a new sync engine.

func (*Engine) GetProgress 

func (e *Engine) GetProgress() *SyncProgress

GetProgress returns current sync progress.

func (*Engine) Pause 

func (e *Engine) Pause() error

Pause pauses the sync engine.

func (*Engine) Resume 

func (e *Engine) Resume() error

Resume resumes a paused sync engine.

func (*Engine) ResumeSession 

func (e *Engine) ResumeSession(ctx context.Context, sessionID string) error

ResumeSession resumes an existing sync session.

func (*Engine) StartNewSession 

func (e *Engine) StartNewSession(ctx context.Context, rootFolderID, destinationPath string) error

StartNewSession starts a new sync session.

func (*Engine) StartNewSessionWithID 

func (e *Engine) StartNewSessionWithID(ctx context.Context, rootFolderID, destinationPath string) (string, error)

StartNewSessionWithID starts a new sync session and returns the session ID.

func (*Engine) Stop 

func (e *Engine) Stop() error

Stop stops the sync engine.

func (*Engine) WaitForCompletion 

func (e *Engine) WaitForCompletion() <-chan struct{}

WaitForCompletion waits until the sync engine completes.

type EngineConfig 

type EngineConfig struct {
	// Folder walker configuration
	WalkerConfig *WalkerConfig

	// Download manager configuration
	DownloadConfig *DownloadManagerConfig

	// Worker pool configuration
	WorkerConfig *WorkerPoolConfig

	// Progress update interval
	ProgressInterval time.Duration

	// Session checkpoint interval
	CheckpointInterval time.Duration

	// Maximum errors before stopping
	MaxErrors int
}

EngineConfig contains configuration for the sync engine.

func DefaultEngineConfig 

func DefaultEngineConfig() *EngineConfig

DefaultEngineConfig returns default engine configuration.

type Event 

type Event struct {
	Timestamp time.Time
	Data      interface{}
	Type      EventType
	Priority  EventPriority
}

Event represents a sync event.

type EventBus 

type EventBus struct {
	// contains filtered or unexported fields
}

EventBus manages event distribution.

func NewEventBus 

func NewEventBus(bufferSize int) *EventBus

NewEventBus creates a new event bus.

func (*EventBus) Close 

func (eb *EventBus) Close()

Close shuts down the event bus.

func (*EventBus) CloseChannel 

func (eb *EventBus) CloseChannel(name string)

CloseChannel closes a named channel.

func (*EventBus) CreateChannel 

func (eb *EventBus) CreateChannel(name string) <-chan Event

CreateChannel creates a named channel for event streaming.

func (*EventBus) Publish 

func (eb *EventBus) Publish(event Event)

Publish publishes an event to all subscribers.

func (*EventBus) PublishFileComplete 

func (eb *EventBus) PublishFileComplete(fileID, fileName string,
	bytesTransferred int64)

PublishFileComplete publishes a file completion event.

func (*EventBus) PublishFileError 

func (eb *EventBus) PublishFileError(fileID, fileName string, err error)

PublishFileError publishes a file error event.

func (*EventBus) PublishFileProgress 

func (eb *EventBus) PublishFileProgress(fileID string, bytesTransferred int64)

PublishFileProgress publishes a file progress event.

func (*EventBus) PublishFileStart 

func (eb *EventBus) PublishFileStart(fileID, fileName, filePath string,
	fileSize int64)

PublishFileStart publishes a file start event.

func (*EventBus) PublishSyncComplete 

func (eb *EventBus) PublishSyncComplete(sessionID string, processedFiles,
	processedBytes int64)

PublishSyncComplete publishes a sync completion event.

func (*EventBus) PublishSyncStart 

func (eb *EventBus) PublishSyncStart(sessionID string, totalFiles,
	totalBytes int64)

PublishSyncStart publishes a sync start event.

func (*EventBus) Subscribe 

func (eb *EventBus) Subscribe(eventType EventType, handler EventHandler,
	filter EventFilter, priority EventPriority)

Subscribe adds a handler for specific event types.

func (*EventBus) SubscribeAll 

func (eb *EventBus) SubscribeAll(handler EventHandler, filter EventFilter,
	priority EventPriority)

SubscribeAll adds a global handler for all events.

type EventFilter 

type EventFilter func(event Event) bool

EventFilter determines if an event should be processed.

type EventHandler 

type EventHandler func(event Event)

EventHandler processes events.

type EventPriority 

type EventPriority int

EventPriority defines the priority of an event. 

const (
	EventPriorityLow EventPriority = iota
	EventPriorityNormal
	EventPriorityHigh
	EventPriorityCritical
)

type EventType 

type EventType int

EventType defines the type of sync event. 

const (
	EventTypeFileStart EventType = iota
	EventTypeFileProgress
	EventTypeFileComplete
	EventTypeFileError
	EventTypeFolderStart
	EventTypeFolderComplete
	EventTypeSyncStart
	EventTypeSyncComplete
	EventTypeSyncPaused
	EventTypeSyncResumed
	EventTypeRateLimit
	EventTypeRetry
)

func (EventType) String 

func (et EventType) String() string

String returns string representation of event type.

type FileEvent 

type FileEvent struct {
	Error            error
	Metadata         map[string]interface{}
	FileID           string
	FileName         string
	FilePath         string
	FileSize         int64
	BytesTransferred int64
}

FileEvent contains data for file-related events.

type FileProgress 

type FileProgress struct {
	StartTime       time.Time
	LastUpdate      time.Time
	FileID          string
	FileName        string
	FilePath        string
	TotalBytes      int64
	BytesDownloaded int64
	Speed           int64
}

FileProgress tracks individual file download progress.

type FolderEvent 

type FolderEvent struct {
	FolderID   string
	FolderName string
	FolderPath string
	FileCount  int
	TotalSize  int64
}

FolderEvent contains data for folder-related events.

type FolderWalker 

type FolderWalker struct {
	// contains filtered or unexported fields
}

FolderWalker implements efficient folder tree traversal.

func NewFolderWalker 

func NewFolderWalker(
	client *api.DriveClient,
	stateManager *state.Manager,
	progressTracker *ProgressTracker,
	logger *logger.Logger,
	config *WalkerConfig,
) (*FolderWalker, error)

NewFolderWalker creates a new folder walker.

func (*FolderWalker) GetStats 

func (fw *FolderWalker) GetStats() *WalkerStats

GetStats returns walker statistics.

func (*FolderWalker) Stop 

func (fw *FolderWalker) Stop()

Stop stops the folder walker.

func (*FolderWalker) Walk 

func (fw *FolderWalker) Walk(ctx context.Context, rootFolderID string, sessionID string) (<-chan *WalkResult, error)

Walk starts walking the folder tree from the given root.

type HandlerInfo 

type HandlerInfo struct {
	Handler  EventHandler
	Filter   EventFilter
	Priority EventPriority
}

HandlerInfo contains handler metadata.

type PriorityQueue 

type PriorityQueue struct {
	// contains filtered or unexported fields
}

PriorityQueue implements a priority queue for download tasks.

func NewPriorityQueue 

func NewPriorityQueue() *PriorityQueue

NewPriorityQueue creates a new priority queue.

func (*PriorityQueue) Len 

func (pq *PriorityQueue) Len() int

Len returns the number of tasks in the queue.

func (*PriorityQueue) Pop 

func (pq *PriorityQueue) Pop() *DownloadTask

Pop removes and returns the highest priority task.

func (*PriorityQueue) Push 

func (pq *PriorityQueue) Push(task *DownloadTask)

Push adds a task to the queue.

type ProgressEvent 

type ProgressEvent struct {
	Timestamp        time.Time
	Error            error
	Context          map[string]interface{}
	SessionID        string
	ItemID           string
	ItemName         string
	ItemPath         string
	ErrorMessage     string
	Type             ProgressEventType
	FilesCompleted   int64
	CurrentSpeed     int64
	AverageSpeed     int64
	RemainingTime    time.Duration
	TotalFiles       int64
	TotalBytes       int64
	BytesTransferred int64
}

ProgressEvent represents a progress update event.

type ProgressEventType 

type ProgressEventType string

ProgressEventType defines types of progress events. 

const (
	// Progress event types as strings for better readability.
	ProgressEventFileStarted     ProgressEventType = "file_started"
	ProgressEventFileProgress    ProgressEventType = "file_progress"
	ProgressEventFileCompleted   ProgressEventType = "file_completed"
	ProgressEventFileFailed      ProgressEventType = "file_failed"
	ProgressEventFolderStarted   ProgressEventType = "folder_started"
	ProgressEventFolderCompleted ProgressEventType = "folder_completed"
	ProgressEventSessionUpdate   ProgressEventType = "session_update"
	ProgressEventBandwidthUpdate ProgressEventType = "bandwidth_update"
)

type ProgressStats 

type ProgressStats struct {
	StartTime       time.Time
	SessionID       string
	FailedFiles     int64
	RemainingTime   time.Duration
	TotalFiles      int64
	CompletedFiles  int64
	ElapsedTime     time.Duration
	SkippedFiles    int64
	TotalBytes      int64
	CompletedBytes  int64
	CurrentSpeed    int64
	AverageSpeed    int64
	ActiveDownloads int
	BandwidthLimit  int64
}

ProgressStats contains current progress statistics.

func (*ProgressStats) BytesProgress 

func (ps *ProgressStats) BytesProgress() float64

BytesProgress returns bytes completion percentage.

func (*ProgressStats) Progress 

func (ps *ProgressStats) Progress() float64

Progress returns completion percentage.

type ProgressTracker 

type ProgressTracker struct {
	// contains filtered or unexported fields
}

ProgressTracker tracks sync progress and emits events.

func NewProgressTracker 

func NewProgressTracker(sessionID string) *ProgressTracker

NewProgressTracker creates a new progress tracker.

func (*ProgressTracker) CheckBandwidthLimit 

func (pt *ProgressTracker) CheckBandwidthLimit(ctx context.Context, bytesRequested int64) error

CheckBandwidthLimit checks if we're within bandwidth limits.

func (*ProgressTracker) FileCompleted 

func (pt *ProgressTracker) FileCompleted(fileID string)

FileCompleted notifies that a file download completed.

func (*ProgressTracker) FileFailed 

func (pt *ProgressTracker) FileFailed(fileID string, err error)

FileFailed notifies that a file download failed.

func (*ProgressTracker) FileProgress 

func (pt *ProgressTracker) FileProgress(fileID string, bytesDownloaded int64)

FileProgress updates file download progress.

func (*ProgressTracker) FileSkipped 

func (pt *ProgressTracker) FileSkipped(fileID, fileName, filePath string, reason string)

FileSkipped notifies that a file was skipped.

func (*ProgressTracker) FileStarted 

func (pt *ProgressTracker) FileStarted(fileID, fileName, filePath string, totalBytes int64)

FileStarted notifies that a file download has started.

func (*ProgressTracker) FolderCompleted 

func (pt *ProgressTracker) FolderCompleted(folderID, folderName, folderPath string, fileCount int64)

FolderCompleted notifies that folder scanning completed.

func (*ProgressTracker) FolderStarted 

func (pt *ProgressTracker) FolderStarted(folderID, folderName, folderPath string)

FolderStarted notifies that folder scanning started.

func (*ProgressTracker) GetStats 

func (pt *ProgressTracker) GetStats() *ProgressStats

GetStats returns current progress statistics.

func (*ProgressTracker) OnEvent 

func (pt *ProgressTracker) OnEvent(handler func(event *ProgressEvent))

OnEvent registers an event handler.

func (*ProgressTracker) SetBandwidthLimit 

func (pt *ProgressTracker) SetBandwidthLimit(bytesPerSecond int64)

SetBandwidthLimit sets the bandwidth limit in bytes per second.

func (*ProgressTracker) SetTotals 

func (pt *ProgressTracker) SetTotals(totalFiles, totalBytes int64)

SetTotals sets the total files and bytes for the session.

type RetryEvent 

type RetryEvent struct {
	NextRetryAt time.Time
	Error       error
	FileID      string
	FileName    string
	RetryCount  int
	MaxRetries  int
}

RetryEvent contains data for retry events.

type SyncEvent 

type SyncEvent struct {
	StartTime      time.Time
	SessionID      string
	Message        string
	TotalFiles     int64
	TotalBytes     int64
	ProcessedFiles int64
	ProcessedBytes int64
}

SyncEvent contains data for sync-related events.

type SyncProgress 

type SyncProgress struct {
	StartTime       time.Time
	SessionID       string
	Status          string
	SkippedFiles    int64
	RemainingTime   time.Duration
	TotalFiles      int64
	CompletedFiles  int64
	FailedFiles     int64
	ElapsedTime     time.Duration
	TotalBytes      int64
	CompletedBytes  int64
	CurrentSpeed    int64
	AverageSpeed    int64
	FoldersScanned  int64
	ActiveDownloads int64
	QueuedDownloads int
}

SyncProgress represents the current sync progress.

type TaskResult 

type TaskResult struct {
	Error        error
	Task         *DownloadTask
	BytesWritten int64
	Duration     time.Duration
	WorkerID     int
	Success      bool
}

TaskResult represents the result of a download task.

type TraversalStrategy 

type TraversalStrategy int

TraversalStrategy defines the folder traversal strategy. 

const (
	// TraversalBFS performs breadth-first search traversal.
	TraversalBFS TraversalStrategy = iota

	// TraversalDFS performs depth-first search traversal.
	TraversalDFS
)

type WalkResult 

type WalkResult struct {
	Error      error
	Folder     *state.Folder
	SkipReason string
	Files      []*state.File
	Depth      int
	IsSkipped  bool
}

WalkResult represents a folder walk result.

type WalkerConfig 

type WalkerConfig struct {
	IncludePatterns   []string
	ExcludePatterns   []string
	Strategy          TraversalStrategy
	MaxDepth          int
	Concurrency       int
	ChannelBufferSize int
	FollowShortcuts   bool
}

WalkerConfig contains configuration for the folder walker.

func DefaultWalkerConfig 

func DefaultWalkerConfig() *WalkerConfig

DefaultWalkerConfig returns default walker configuration.

type WalkerStats 

type WalkerStats struct {
	FoldersScanned int64
	FilesFound     int64
	TotalSize      int64
	ErrorCount     int
}

WalkerStats contains walker statistics.

type Worker 

type Worker struct {
	// contains filtered or unexported fields
}

Worker represents a download worker.

type WorkerPool 

type WorkerPool struct {
	// contains filtered or unexported fields
}

WorkerPool manages concurrent download workers.

func NewWorkerPool 

func NewWorkerPool(
	client *api.DriveClient,
	stateManager *state.Manager,
	progressTracker *ProgressTracker,
	errorHandler *errors.Handler,
	logger *logger.Logger,
	config *WorkerPoolConfig,
) *WorkerPool

NewWorkerPool creates a new worker pool.

func (*WorkerPool) GetStats 

func (wp *WorkerPool) GetStats() *WorkerPoolStats

GetStats returns worker pool statistics.

func (*WorkerPool) SetDownloadManager 

func (wp *WorkerPool) SetDownloadManager(dm *DownloadManager)

SetDownloadManager sets the download manager reference.

func (*WorkerPool) Start 

func (wp *WorkerPool) Start(ctx context.Context) error

Start starts the worker pool.

func (*WorkerPool) Stop 

func (wp *WorkerPool) Stop() error

Stop stops the worker pool gracefully.

func (*WorkerPool) SubmitTask 

func (wp *WorkerPool) SubmitTask(file *state.File, priority int) error

SubmitTask submits a download task to the pool.

type WorkerPoolConfig 

type WorkerPoolConfig struct {
	WorkerCount     int
	MaxRetries      int
	ShutdownTimeout time.Duration
}

WorkerPoolConfig contains configuration for the worker pool.

func DefaultWorkerPoolConfig 

func DefaultWorkerPoolConfig() *WorkerPoolConfig

DefaultWorkerPoolConfig returns default configuration.

type WorkerPoolStats 

type WorkerPoolStats struct {
	WorkerCount     int
	ActiveWorkers   int
	QueuedTasks     int
	TasksProcessed  int64
	TasksSucceeded  int64
	TasksFailed     int64
	BytesDownloaded int64
}

WorkerPoolStats contains worker pool statistics.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.