593 lines
16 KiB
Go
593 lines
16 KiB
Go
// Package syncer mirrors files between S3 objects and a local directory.
|
|
package syncer
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"path"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/aws/aws-sdk-go-v2/aws"
|
|
"github.com/aws/aws-sdk-go-v2/service/s3"
|
|
"github.com/aws/aws-sdk-go-v2/service/s3/types"
|
|
)
|
|
|
|
const (
|
|
stateDirName = ".s3watch"
|
|
stateFileName = "state.json"
|
|
tempFileGlob = ".s3watch-*"
|
|
)
|
|
|
|
// Config controls S3/local sync behavior.
|
|
type Config struct {
|
|
Bucket string
|
|
Prefix string
|
|
Dir string
|
|
Prune bool
|
|
}
|
|
|
|
// Result describes changes applied during one sync cycle.
|
|
type Result struct {
|
|
Downloaded int
|
|
Uploaded int
|
|
Updated int
|
|
Deleted int
|
|
Unchanged int
|
|
}
|
|
|
|
// Changed reports whether a sync cycle changed either side.
|
|
func (r Result) Changed() bool {
|
|
return r.Downloaded > 0 || r.Uploaded > 0 || r.Updated > 0 || r.Deleted > 0
|
|
}
|
|
|
|
// S3API is the subset of S3 operations required by Syncer.
|
|
type S3API interface {
|
|
ListObjectsV2(context.Context, *s3.ListObjectsV2Input, ...func(*s3.Options)) (*s3.ListObjectsV2Output, error)
|
|
GetObject(context.Context, *s3.GetObjectInput, ...func(*s3.Options)) (*s3.GetObjectOutput, error)
|
|
PutObject(context.Context, *s3.PutObjectInput, ...func(*s3.Options)) (*s3.PutObjectOutput, error)
|
|
HeadObject(context.Context, *s3.HeadObjectInput, ...func(*s3.Options)) (*s3.HeadObjectOutput, error)
|
|
DeleteObject(context.Context, *s3.DeleteObjectInput, ...func(*s3.Options)) (*s3.DeleteObjectOutput, error)
|
|
}
|
|
|
|
// Syncer synchronizes S3 objects and local files.
|
|
type Syncer struct {
|
|
client S3API
|
|
cfg Config
|
|
}
|
|
|
|
// New creates a Syncer.
|
|
func New(client S3API, cfg Config) *Syncer {
|
|
return &Syncer{
|
|
client: client,
|
|
cfg: cfg,
|
|
}
|
|
}
|
|
|
|
// Sync reconciles configured S3 objects and the configured local directory.
|
|
func (s *Syncer) Sync(ctx context.Context) (Result, error) {
|
|
if s.client == nil {
|
|
return Result{}, fmt.Errorf("s3 client is required")
|
|
}
|
|
if s.cfg.Bucket == "" {
|
|
return Result{}, fmt.Errorf("bucket is required")
|
|
}
|
|
if s.cfg.Dir == "" {
|
|
return Result{}, fmt.Errorf("dir is required")
|
|
}
|
|
|
|
if err := os.MkdirAll(s.cfg.Dir, 0o755); err != nil {
|
|
return Result{}, fmt.Errorf("creating sync directory: %w", err)
|
|
}
|
|
|
|
previous, err := loadState(s.statePath())
|
|
if err != nil {
|
|
return Result{}, err
|
|
}
|
|
|
|
remoteFiles, err := s.listRemoteFiles(ctx)
|
|
if err != nil {
|
|
return Result{}, err
|
|
}
|
|
|
|
localFiles, err := scanLocalFiles(s.cfg.Dir)
|
|
if err != nil {
|
|
return Result{}, err
|
|
}
|
|
|
|
result := Result{}
|
|
for _, rel := range unionKeys(localFiles, remoteFiles, previous.Files) {
|
|
local, localOK := localFiles[rel]
|
|
remote, remoteOK := remoteFiles[rel]
|
|
prior, priorOK := previous.Files[rel]
|
|
|
|
switch {
|
|
case localOK && remoteOK:
|
|
changed, err := s.syncExisting(ctx, rel, local, remote, localFiles, remoteFiles)
|
|
if err != nil {
|
|
return Result{}, err
|
|
}
|
|
if changed {
|
|
result.Updated++
|
|
} else {
|
|
result.Unchanged++
|
|
}
|
|
case localOK:
|
|
if s.shouldDeleteLocal(local, prior, priorOK) {
|
|
if err := os.Remove(local.Path); err != nil {
|
|
return Result{}, fmt.Errorf("deleting local %s: %w", local.Path, err)
|
|
}
|
|
delete(localFiles, rel)
|
|
result.Deleted++
|
|
continue
|
|
}
|
|
if err := s.uploadFile(ctx, rel, local, localFiles, remoteFiles); err != nil {
|
|
return Result{}, err
|
|
}
|
|
result.Uploaded++
|
|
case remoteOK:
|
|
if s.shouldDeleteRemote(remote, prior, priorOK) {
|
|
if err := s.deleteRemote(ctx, rel); err != nil {
|
|
return Result{}, err
|
|
}
|
|
delete(remoteFiles, rel)
|
|
result.Deleted++
|
|
continue
|
|
}
|
|
if err := s.downloadFile(ctx, rel, remote, localFiles); err != nil {
|
|
return Result{}, err
|
|
}
|
|
result.Downloaded++
|
|
}
|
|
}
|
|
|
|
if err := saveState(s.statePath(), buildState(localFiles, remoteFiles)); err != nil {
|
|
return Result{}, err
|
|
}
|
|
|
|
return result, nil
|
|
}
|
|
|
|
func (s *Syncer) syncExisting(ctx context.Context, rel string, local localFile, remote remoteFile, localFiles map[string]localFile, remoteFiles map[string]remoteFile) (bool, error) {
|
|
if local.Snapshot.Equal(remote.Snapshot) {
|
|
return false, nil
|
|
}
|
|
|
|
if local.Snapshot.ModTimeUnix >= remote.Snapshot.ModTimeUnix {
|
|
if err := s.uploadFile(ctx, rel, local, localFiles, remoteFiles); err != nil {
|
|
return false, err
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
if err := s.downloadFile(ctx, rel, remote, localFiles); err != nil {
|
|
return false, err
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
func (s *Syncer) shouldDeleteLocal(local localFile, prior stateEntry, priorOK bool) bool {
|
|
return s.cfg.Prune && priorOK && prior.Remote != nil && prior.Local != nil && local.Snapshot.Equal(*prior.Local)
|
|
}
|
|
|
|
func (s *Syncer) shouldDeleteRemote(remote remoteFile, prior stateEntry, priorOK bool) bool {
|
|
return s.cfg.Prune && priorOK && prior.Local != nil && prior.Remote != nil && remote.Snapshot.Equal(*prior.Remote)
|
|
}
|
|
|
|
func (s *Syncer) listRemoteFiles(ctx context.Context) (map[string]remoteFile, error) {
|
|
files := make(map[string]remoteFile)
|
|
var token *string
|
|
|
|
prefix := normalizePrefix(s.cfg.Prefix)
|
|
for {
|
|
output, err := s.client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
|
|
Bucket: aws.String(s.cfg.Bucket),
|
|
Prefix: aws.String(prefix),
|
|
ContinuationToken: token,
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("listing s3://%s/%s: %w", s.cfg.Bucket, prefix, err)
|
|
}
|
|
|
|
for _, object := range output.Contents {
|
|
if object.Key == nil || strings.HasSuffix(*object.Key, "/") {
|
|
continue
|
|
}
|
|
rel, err := relativeKey(*object.Key, s.cfg.Prefix)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if _, err := safeLocalPath(s.cfg.Dir, rel); err != nil {
|
|
return nil, err
|
|
}
|
|
files[rel] = remoteFile{
|
|
Key: *object.Key,
|
|
Snapshot: snapshotFromObject(object),
|
|
}
|
|
}
|
|
|
|
if output.IsTruncated == nil || !*output.IsTruncated {
|
|
return files, nil
|
|
}
|
|
if output.NextContinuationToken == nil {
|
|
return nil, fmt.Errorf("listing s3://%s/%s: truncated response missing continuation token", s.cfg.Bucket, prefix)
|
|
}
|
|
token = output.NextContinuationToken
|
|
}
|
|
}
|
|
|
|
func (s *Syncer) downloadFile(ctx context.Context, rel string, remote remoteFile, localFiles map[string]localFile) error {
|
|
localPath, err := safeLocalPath(s.cfg.Dir, rel)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := os.MkdirAll(filepath.Dir(localPath), 0o755); err != nil {
|
|
return fmt.Errorf("creating parent directory for %s: %w", localPath, err)
|
|
}
|
|
|
|
output, err := s.client.GetObject(ctx, &s3.GetObjectInput{
|
|
Bucket: aws.String(s.cfg.Bucket),
|
|
Key: aws.String(remote.Key),
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("getting s3://%s/%s: %w", s.cfg.Bucket, remote.Key, err)
|
|
}
|
|
defer output.Body.Close()
|
|
|
|
temp, err := os.CreateTemp(filepath.Dir(localPath), tempFileGlob)
|
|
if err != nil {
|
|
return fmt.Errorf("creating temporary file for %s: %w", localPath, err)
|
|
}
|
|
tempPath := temp.Name()
|
|
removeTemp := true
|
|
defer func() {
|
|
if removeTemp {
|
|
_ = os.Remove(tempPath)
|
|
}
|
|
}()
|
|
|
|
if _, err := io.Copy(temp, output.Body); err != nil {
|
|
_ = temp.Close()
|
|
return fmt.Errorf("writing temporary file for %s: %w", localPath, err)
|
|
}
|
|
if err := temp.Close(); err != nil {
|
|
return fmt.Errorf("closing temporary file for %s: %w", localPath, err)
|
|
}
|
|
|
|
if err := os.Chtimes(tempPath, unixTime(remote.Snapshot.ModTimeUnix), unixTime(remote.Snapshot.ModTimeUnix)); err != nil {
|
|
return fmt.Errorf("setting timestamp on %s: %w", tempPath, err)
|
|
}
|
|
|
|
if err := os.Rename(tempPath, localPath); err != nil {
|
|
return fmt.Errorf("replacing %s: %w", localPath, err)
|
|
}
|
|
removeTemp = false
|
|
|
|
localFiles[rel] = localFile{
|
|
Path: localPath,
|
|
Snapshot: remote.Snapshot,
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Syncer) uploadFile(ctx context.Context, rel string, local localFile, localFiles map[string]localFile, remoteFiles map[string]remoteFile) error {
|
|
file, err := os.Open(local.Path)
|
|
if err != nil {
|
|
return fmt.Errorf("opening local %s: %w", local.Path, err)
|
|
}
|
|
defer file.Close()
|
|
|
|
key := s.remoteKey(rel)
|
|
if _, err := s.client.PutObject(ctx, &s3.PutObjectInput{
|
|
Bucket: aws.String(s.cfg.Bucket),
|
|
Key: aws.String(key),
|
|
Body: file,
|
|
ContentLength: aws.Int64(local.Snapshot.Size),
|
|
}); err != nil {
|
|
return fmt.Errorf("putting s3://%s/%s: %w", s.cfg.Bucket, key, err)
|
|
}
|
|
|
|
head, err := s.client.HeadObject(ctx, &s3.HeadObjectInput{
|
|
Bucket: aws.String(s.cfg.Bucket),
|
|
Key: aws.String(key),
|
|
})
|
|
if err != nil {
|
|
return fmt.Errorf("heading uploaded s3://%s/%s: %w", s.cfg.Bucket, key, err)
|
|
}
|
|
|
|
remoteSnapshot := snapshotFromHead(head, local.Snapshot)
|
|
if err := os.Chtimes(local.Path, unixTime(remoteSnapshot.ModTimeUnix), unixTime(remoteSnapshot.ModTimeUnix)); err != nil {
|
|
return fmt.Errorf("setting timestamp on %s: %w", local.Path, err)
|
|
}
|
|
|
|
localFiles[rel] = localFile{
|
|
Path: local.Path,
|
|
Snapshot: remoteSnapshot,
|
|
}
|
|
remoteFiles[rel] = remoteFile{
|
|
Key: key,
|
|
Snapshot: remoteSnapshot,
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Syncer) deleteRemote(ctx context.Context, rel string) error {
|
|
key := s.remoteKey(rel)
|
|
if _, err := s.client.DeleteObject(ctx, &s3.DeleteObjectInput{
|
|
Bucket: aws.String(s.cfg.Bucket),
|
|
Key: aws.String(key),
|
|
}); err != nil {
|
|
return fmt.Errorf("deleting s3://%s/%s: %w", s.cfg.Bucket, key, err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *Syncer) remoteKey(rel string) string {
|
|
prefix := normalizePrefix(s.cfg.Prefix)
|
|
return prefix + strings.TrimLeft(path.Clean(rel), "/")
|
|
}
|
|
|
|
func (s *Syncer) statePath() string {
|
|
return filepath.Join(s.cfg.Dir, stateDirName, stateFileName)
|
|
}
|
|
|
|
type localFile struct {
|
|
Path string
|
|
Snapshot fileSnapshot
|
|
}
|
|
|
|
type remoteFile struct {
|
|
Key string
|
|
Snapshot fileSnapshot
|
|
}
|
|
|
|
type fileSnapshot struct {
|
|
Size int64 `json:"size"`
|
|
ModTimeUnix int64 `json:"mod_time_unix"`
|
|
}
|
|
|
|
// Equal reports whether two snapshots describe the same file state.
|
|
func (s fileSnapshot) Equal(other fileSnapshot) bool {
|
|
return s.Size == other.Size && s.ModTimeUnix == other.ModTimeUnix
|
|
}
|
|
|
|
type syncState struct {
|
|
Version int `json:"version"`
|
|
Files map[string]stateEntry `json:"files"`
|
|
}
|
|
|
|
type stateEntry struct {
|
|
Local *fileSnapshot `json:"local,omitempty"`
|
|
Remote *fileSnapshot `json:"remote,omitempty"`
|
|
}
|
|
|
|
func loadState(path string) (syncState, error) {
|
|
state := syncState{
|
|
Version: 1,
|
|
Files: map[string]stateEntry{},
|
|
}
|
|
|
|
file, err := os.Open(path)
|
|
if os.IsNotExist(err) {
|
|
return state, nil
|
|
}
|
|
if err != nil {
|
|
return syncState{}, fmt.Errorf("opening sync state: %w", err)
|
|
}
|
|
defer file.Close()
|
|
|
|
if err := json.NewDecoder(file).Decode(&state); err != nil {
|
|
return syncState{}, fmt.Errorf("decoding sync state: %w", err)
|
|
}
|
|
if state.Files == nil {
|
|
state.Files = map[string]stateEntry{}
|
|
}
|
|
return state, nil
|
|
}
|
|
|
|
func saveState(path string, state syncState) error {
|
|
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
|
return fmt.Errorf("creating sync state directory: %w", err)
|
|
}
|
|
|
|
temp, err := os.CreateTemp(filepath.Dir(path), tempFileGlob)
|
|
if err != nil {
|
|
return fmt.Errorf("creating sync state temporary file: %w", err)
|
|
}
|
|
tempPath := temp.Name()
|
|
removeTemp := true
|
|
defer func() {
|
|
if removeTemp {
|
|
_ = os.Remove(tempPath)
|
|
}
|
|
}()
|
|
|
|
encoder := json.NewEncoder(temp)
|
|
encoder.SetIndent("", " ")
|
|
if err := encoder.Encode(state); err != nil {
|
|
_ = temp.Close()
|
|
return fmt.Errorf("encoding sync state: %w", err)
|
|
}
|
|
if err := temp.Close(); err != nil {
|
|
return fmt.Errorf("closing sync state temporary file: %w", err)
|
|
}
|
|
if err := os.Rename(tempPath, path); err != nil {
|
|
return fmt.Errorf("replacing sync state: %w", err)
|
|
}
|
|
removeTemp = false
|
|
return nil
|
|
}
|
|
|
|
func scanLocalFiles(root string) (map[string]localFile, error) {
|
|
files := make(map[string]localFile)
|
|
rootAbs, err := filepath.Abs(root)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("resolving root directory: %w", err)
|
|
}
|
|
stateDir := filepath.Join(rootAbs, stateDirName)
|
|
|
|
if err := filepath.WalkDir(rootAbs, func(current string, entry os.DirEntry, err error) error {
|
|
if err != nil {
|
|
return fmt.Errorf("walking %s: %w", current, err)
|
|
}
|
|
if current == stateDir && entry.IsDir() {
|
|
return filepath.SkipDir
|
|
}
|
|
if entry.IsDir() {
|
|
return nil
|
|
}
|
|
if entry.Type()&os.ModeSymlink != 0 {
|
|
return nil
|
|
}
|
|
if strings.HasPrefix(entry.Name(), strings.TrimSuffix(tempFileGlob, "*")) {
|
|
return nil
|
|
}
|
|
|
|
info, err := entry.Info()
|
|
if err != nil {
|
|
return fmt.Errorf("statting %s: %w", current, err)
|
|
}
|
|
rel, err := filepath.Rel(rootAbs, current)
|
|
if err != nil {
|
|
return fmt.Errorf("relativizing %s: %w", current, err)
|
|
}
|
|
rel = filepath.ToSlash(rel)
|
|
files[rel] = localFile{
|
|
Path: current,
|
|
Snapshot: fileSnapshot{
|
|
Size: info.Size(),
|
|
ModTimeUnix: info.ModTime().Unix(),
|
|
},
|
|
}
|
|
return nil
|
|
}); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return files, nil
|
|
}
|
|
|
|
func buildState(localFiles map[string]localFile, remoteFiles map[string]remoteFile) syncState {
|
|
state := syncState{
|
|
Version: 1,
|
|
Files: make(map[string]stateEntry),
|
|
}
|
|
for _, rel := range unionKeys(localFiles, remoteFiles, nil) {
|
|
entry := stateEntry{}
|
|
if local, ok := localFiles[rel]; ok {
|
|
snapshot := local.Snapshot
|
|
entry.Local = &snapshot
|
|
}
|
|
if remote, ok := remoteFiles[rel]; ok {
|
|
snapshot := remote.Snapshot
|
|
entry.Remote = &snapshot
|
|
}
|
|
state.Files[rel] = entry
|
|
}
|
|
return state
|
|
}
|
|
|
|
func unionKeys(localFiles map[string]localFile, remoteFiles map[string]remoteFile, previous map[string]stateEntry) []string {
|
|
keys := make(map[string]struct{}, len(localFiles)+len(remoteFiles)+len(previous))
|
|
for key := range localFiles {
|
|
keys[key] = struct{}{}
|
|
}
|
|
for key := range remoteFiles {
|
|
keys[key] = struct{}{}
|
|
}
|
|
for key := range previous {
|
|
keys[key] = struct{}{}
|
|
}
|
|
|
|
out := make([]string, 0, len(keys))
|
|
for key := range keys {
|
|
out = append(out, key)
|
|
}
|
|
sort.Strings(out)
|
|
return out
|
|
}
|
|
|
|
func snapshotFromObject(object types.Object) fileSnapshot {
|
|
snapshot := fileSnapshot{}
|
|
if object.Size != nil {
|
|
snapshot.Size = *object.Size
|
|
}
|
|
if object.LastModified != nil {
|
|
snapshot.ModTimeUnix = object.LastModified.Unix()
|
|
}
|
|
return snapshot
|
|
}
|
|
|
|
func snapshotFromHead(output *s3.HeadObjectOutput, fallback fileSnapshot) fileSnapshot {
|
|
snapshot := fallback
|
|
if output.ContentLength != nil {
|
|
snapshot.Size = *output.ContentLength
|
|
}
|
|
if output.LastModified != nil {
|
|
snapshot.ModTimeUnix = output.LastModified.Unix()
|
|
}
|
|
return snapshot
|
|
}
|
|
|
|
func normalizePrefix(prefix string) string {
|
|
prefix = strings.Trim(prefix, "/")
|
|
if prefix == "" {
|
|
return ""
|
|
}
|
|
return strings.TrimSuffix(prefix, "/") + "/"
|
|
}
|
|
|
|
func relativeKey(key string, prefix string) (string, error) {
|
|
normalizedPrefix := normalizePrefix(prefix)
|
|
if normalizedPrefix != "" {
|
|
if !strings.HasPrefix(key, normalizedPrefix) {
|
|
return "", fmt.Errorf("s3 key %q does not match prefix %q", key, normalizedPrefix)
|
|
}
|
|
key = strings.TrimPrefix(key, normalizedPrefix)
|
|
}
|
|
if key == "" {
|
|
return "", fmt.Errorf("s3 key resolves to empty local path")
|
|
}
|
|
return key, nil
|
|
}
|
|
|
|
func safeLocalPath(root string, rel string) (string, error) {
|
|
if rel == "" {
|
|
return "", fmt.Errorf("empty relative path")
|
|
}
|
|
if strings.HasPrefix(rel, "/") {
|
|
return "", fmt.Errorf("unsafe absolute s3 key %q", rel)
|
|
}
|
|
|
|
clean := path.Clean(rel)
|
|
for _, segment := range strings.Split(clean, "/") {
|
|
if segment == "." || segment == ".." || segment == "" {
|
|
return "", fmt.Errorf("unsafe s3 key %q", rel)
|
|
}
|
|
}
|
|
|
|
localPath := filepath.Join(root, filepath.FromSlash(clean))
|
|
rootAbs, err := filepath.Abs(root)
|
|
if err != nil {
|
|
return "", fmt.Errorf("resolving root directory: %w", err)
|
|
}
|
|
localAbs, err := filepath.Abs(localPath)
|
|
if err != nil {
|
|
return "", fmt.Errorf("resolving local path: %w", err)
|
|
}
|
|
if localAbs != rootAbs && !strings.HasPrefix(localAbs, rootAbs+string(os.PathSeparator)) {
|
|
return "", fmt.Errorf("unsafe s3 key %q", rel)
|
|
}
|
|
|
|
return localAbs, nil
|
|
}
|
|
|
|
func unixTime(seconds int64) time.Time {
|
|
return time.Unix(seconds, 0)
|
|
}
|