commit 36ad6ab38bcb807e0f88e79f80eba6e0f0a9a758 Author: Glauber Ferreira Date: Wed Sep 16 10:27:58 2026 -0300 primeiro commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a14841b --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +bin +.codex +.agents diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..46c08f7 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,62 @@ +## Code Style + +- Follow standard Go conventions: `gofmt`, `goimports`, and `golangci-lint`. +- Keep packages small and focused. Name them short, lowercase, no underscores. +- Avoid unnecessary abstractions — prefer simple, readable code over clever code. +- Define interfaces at the point of use, not the point of implementation. Keep them small. + +## Error Handling + +- Always handle errors explicitly — never ignore them with `_`. +- Wrap errors with context: `fmt.Errorf("doing X: %w", err)`. +- Use `errors.Is` / `errors.As` for matching, not string comparison. +- Reserve `panic` for unrecoverable programmer mistakes, not runtime errors. + +## Concurrency + +- Pass `context.Context` as the first argument to any blocking or I/O function. +- Never store a `Context` in a struct. Never use `context.Background()` deep in business logic. +- Always manage goroutine lifecycles — use `errgroup`, `sync.WaitGroup`, or channels. +- Protect shared state with mutexes or by confining it to a single goroutine. + +## Testing + +- Write table-driven tests using `t.Run` for clarity and coverage. +- Use the standard `testing` package. Reach for `testify` only when it genuinely reduces noise. +- Prefer real implementations over mocks where feasible (e.g., `httptest`, in-memory stores). +- Keep tests close to the code they test; use `_test` packages for black-box testing. +- Benchmark with `testing.B` before optimizing anything. + +## Dependencies & Modules + +- Keep `go.mod` tidy — run `go mod tidy` before committing. +- Minimize external dependencies; prefer the standard library. +- Pin versions explicitly and review dependency updates carefully. + +## General + +- Measure before optimizing. Use `pprof` for profiling. +- Log at boundaries (entry/exit of services), not inside every function. +- Prefer explicit over implicit — avoid `init()` and global state. + +## Documentation + +- Document every exported type, function, method, and constant — no exceptions. +- Follow Go doc conventions: start the comment with the name of the thing being documented. +- Package-level comments should explain purpose and usage, not implementation details. + +## Testing (expanded) + +- Write tests as you write code — not after, not when asked. Tests are not optional. +- Every exported function and critical internal path must have at least one test. +- Cover edge cases, error paths, and boundary conditions — not just the happy path. +- Use `t.Helper()` in shared test utilities to keep failure output pointing at the call site. +- Integration tests live in `test/` or behind a build tag (e.g., `//go:build integration`). + +## Security + +- Never log secrets, tokens, passwords, or PII — scrub them before logging. +- Validate and sanitize all external input; never trust data from outside the process boundary. +- Use `crypto/rand` for all randomness that touches security; never `math/rand`. +- Scan dependencies for known CVEs regularly (e.g., `govulncheck`). +- Set strict timeouts on all outbound HTTP clients — never use the default zero-timeout client. diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..f9b935d --- /dev/null +++ b/Makefile @@ -0,0 +1,18 @@ +.PHONY: fmt test build build-linux-arm64 tidy + +fmt: + gofmt -w ./cmd ./internal + +test: + go test ./... + +build: + mkdir -p bin + env CGO_ENABLED=0 go build -buildvcs=false -o bin/s3watch ./cmd/s3watch + +build-linux-arm64: + mkdir -p bin + env CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -buildvcs=false -o bin/s3watch-linux-arm64 ./cmd/s3watch + +tidy: + go mod tidy diff --git a/README.md b/README.md new file mode 100644 index 0000000..540154a --- /dev/null +++ b/README.md @@ -0,0 +1,82 @@ +# s3watch + +`s3watch` is a Go daemon that periodically keeps an S3 bucket or prefix and a +local directory in sync. When a sync cycle downloads, uploads, updates, or +deletes files, it runs an optional hook script. + +The daemon uses the AWS SDK directly. It does not shell out to `aws s3 sync` or +any other external sync tool. + +## Usage + +```sh +s3watch -bucket my-bucket -prefix app/config -dir /srv/config -hook /usr/local/bin/reload-app +``` + +You can also use a JSON config file: + +```sh +s3watch -config examples/s3watch.json +``` + +Useful flags: + +- `-config examples/s3watch.json`: loads daemon settings from JSON. +- `-interval 1m`: controls the sync period. +- `-once`: runs a single sync cycle and exits. +- `-prune`: propagates deletes for files that were previously synced and are unchanged on the remaining side. +- `-region us-east-1`: overrides the AWS SDK's default region resolution. +- `-key-id KEY`: uses explicit S3 access key credentials. +- `-application-key SECRET`: uses explicit S3 secret or application key credentials. +- `-endpoint-url https://s3.example.com`: uses a custom S3-compatible endpoint. +- `-http-timeout 30s`: sets a timeout for AWS HTTP requests. + +Flags override values loaded from `-config`. + +Example config: + +```json +{ + "bucket": "my-application-config", + "prefix": "production/web", + "dir": "/srv/myapp/config", + "hook": "/usr/local/bin/reload-myapp", + "region": "us-east-1", + "key_id": "REPLACE_WITH_KEY_ID", + "application_key": "REPLACE_WITH_APPLICATION_KEY", + "endpoint_url": "https://s3.us-east-1.amazonaws.com", + "interval": "1m", + "http_timeout": "30s", + "once": false, + "prune": true +} +``` + +The hook runs with the synced directory as its working directory and receives: + +- `S3WATCH_CHANGED` +- `S3WATCH_DOWNLOADED` +- `S3WATCH_UPLOADED` +- `S3WATCH_UPDATED` +- `S3WATCH_DELETED` +- `S3WATCH_UNCHANGED` + +## Sync Semantics + +Local-only files are uploaded to S3. Remote-only objects are downloaded locally. +When a file exists on both sides and differs, the side with the newer modified +time wins. + +With `prune` disabled, missing files are treated as new files on the side where +they still exist. With `prune` enabled, `s3watch` uses local state in +`.s3watch/state.json` to propagate deletes only when the remaining copy is +unchanged since the last successful sync. + +## Credentials + +AWS credentials are resolved by the standard AWS SDK chain, including +environment variables, shared config files, web identity, and instance or task +roles. + +If `key_id` and `application_key` are set, those static credentials override the +standard AWS SDK credential chain. They must be provided together. diff --git a/cmd/s3watch/main.go b/cmd/s3watch/main.go new file mode 100644 index 0000000..f22937e --- /dev/null +++ b/cmd/s3watch/main.go @@ -0,0 +1,253 @@ +// Package main provides the s3watch daemon command. +package main + +import ( + "context" + "encoding/json" + "flag" + "fmt" + "log/slog" + "net/http" + "net/url" + "os" + "os/signal" + "syscall" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/config" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/s3" + + "s3watch/internal/daemon" + "s3watch/internal/hook" + "s3watch/internal/syncer" +) + +func main() { + if err := run(); err != nil { + slog.Error("s3watch failed", "error", err) + os.Exit(1) + } +} + +func run() error { + cfg, err := parseConfig(os.Args[1:]) + if err != nil { + return err + } + if err := cfg.validate(); err != nil { + return err + } + + ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) + defer stop() + + awsOptions := []func(*config.LoadOptions) error{ + config.WithHTTPClient(&http.Client{Timeout: cfg.httpTimeout}), + } + if cfg.region != "" { + awsOptions = append(awsOptions, config.WithRegion(cfg.region)) + } + if cfg.keyID != "" && cfg.applicationKey != "" { + awsOptions = append(awsOptions, config.WithCredentialsProvider( + credentials.NewStaticCredentialsProvider(cfg.keyID, cfg.applicationKey, ""), + )) + } + + awsConfig, err := config.LoadDefaultConfig(ctx, awsOptions...) + if err != nil { + return fmt.Errorf("loading AWS config: %w", err) + } + + s3Client := s3.NewFromConfig(awsConfig, func(options *s3.Options) { + if cfg.endpointURL != "" { + options.BaseEndpoint = aws.String(cfg.endpointURL) + } + }) + syncService := syncer.New(s3Client, syncer.Config{ + Bucket: cfg.bucket, + Prefix: cfg.prefix, + Dir: cfg.dir, + Prune: cfg.prune, + }) + + var runner daemon.HookRunner + if cfg.hook != "" { + runner = hook.NewRunner(cfg.hook, cfg.dir) + } + + return daemon.Run(ctx, daemon.Config{ + Interval: cfg.interval, + Once: cfg.once, + }, syncService, runner, slog.Default()) +} + +type cliConfig struct { + bucket string + prefix string + dir string + hook string + region string + keyID string + applicationKey string + endpointURL string + interval time.Duration + httpTimeout time.Duration + once bool + prune bool +} + +func (c cliConfig) validate() error { + if c.bucket == "" { + return fmt.Errorf("bucket is required") + } + if c.dir == "" { + return fmt.Errorf("dir is required") + } + if c.interval <= 0 { + return fmt.Errorf("interval must be greater than zero") + } + if c.httpTimeout <= 0 { + return fmt.Errorf("http-timeout must be greater than zero") + } + if (c.keyID == "") != (c.applicationKey == "") { + return fmt.Errorf("key-id and application-key must be provided together") + } + if c.endpointURL != "" { + parsed, err := url.Parse(c.endpointURL) + if err != nil { + return fmt.Errorf("parsing endpoint-url: %w", err) + } + if parsed.Scheme == "" || parsed.Host == "" { + return fmt.Errorf("endpoint-url must include scheme and host") + } + } + return nil +} + +func parseConfig(args []string) (cliConfig, error) { + cfg := defaultConfig() + var configPath string + fs := newFlagSet("s3watch", &cfg, &configPath) + if err := fs.Parse(args); err != nil { + return cliConfig{}, err + } + + if configPath == "" { + return cfg, nil + } + + cfg = defaultConfig() + if err := loadConfigFile(configPath, &cfg); err != nil { + return cliConfig{}, err + } + + fs = newFlagSet("s3watch", &cfg, &configPath) + if err := fs.Parse(args); err != nil { + return cliConfig{}, err + } + return cfg, nil +} + +func defaultConfig() cliConfig { + return cliConfig{ + interval: time.Minute, + httpTimeout: 30 * time.Second, + } +} + +func newFlagSet(name string, cfg *cliConfig, configPath *string) *flag.FlagSet { + fs := flag.NewFlagSet(name, flag.ExitOnError) + fs.StringVar(configPath, "config", "", "JSON config file path") + fs.StringVar(&cfg.bucket, "bucket", cfg.bucket, "S3 bucket name to sync from") + fs.StringVar(&cfg.prefix, "prefix", cfg.prefix, "S3 key prefix to sync") + fs.StringVar(&cfg.dir, "dir", cfg.dir, "local directory to sync into") + fs.DurationVar(&cfg.interval, "interval", cfg.interval, "sync interval") + fs.StringVar(&cfg.hook, "hook", cfg.hook, "executable hook script to run after changes") + fs.BoolVar(&cfg.once, "once", cfg.once, "run a single sync cycle and exit") + fs.BoolVar(&cfg.prune, "prune", cfg.prune, "delete local files that are missing from S3") + fs.StringVar(&cfg.region, "region", cfg.region, "AWS region override") + fs.StringVar(&cfg.keyID, "key-id", cfg.keyID, "S3 access key ID override") + fs.StringVar(&cfg.applicationKey, "application-key", cfg.applicationKey, "S3 secret/application key override") + fs.StringVar(&cfg.endpointURL, "endpoint-url", cfg.endpointURL, "S3 endpoint URL override") + fs.DurationVar(&cfg.httpTimeout, "http-timeout", cfg.httpTimeout, "timeout for AWS HTTP requests") + return fs +} + +func loadConfigFile(path string, cfg *cliConfig) error { + file, err := os.Open(path) + if err != nil { + return fmt.Errorf("opening config file: %w", err) + } + defer file.Close() + + var disk configFile + decoder := json.NewDecoder(file) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&disk); err != nil { + return fmt.Errorf("decoding config file: %w", err) + } + + if disk.Bucket != nil { + cfg.bucket = *disk.Bucket + } + if disk.Prefix != nil { + cfg.prefix = *disk.Prefix + } + if disk.Dir != nil { + cfg.dir = *disk.Dir + } + if disk.Hook != nil { + cfg.hook = *disk.Hook + } + if disk.Region != nil { + cfg.region = *disk.Region + } + if disk.KeyID != nil { + cfg.keyID = *disk.KeyID + } + if disk.ApplicationKey != nil { + cfg.applicationKey = *disk.ApplicationKey + } + if disk.EndpointURL != nil { + cfg.endpointURL = *disk.EndpointURL + } + if disk.Interval != nil { + duration, err := time.ParseDuration(*disk.Interval) + if err != nil { + return fmt.Errorf("parsing interval: %w", err) + } + cfg.interval = duration + } + if disk.HTTPTimeout != nil { + duration, err := time.ParseDuration(*disk.HTTPTimeout) + if err != nil { + return fmt.Errorf("parsing http_timeout: %w", err) + } + cfg.httpTimeout = duration + } + if disk.Once != nil { + cfg.once = *disk.Once + } + if disk.Prune != nil { + cfg.prune = *disk.Prune + } + + return nil +} + +type configFile struct { + Bucket *string `json:"bucket"` + Prefix *string `json:"prefix"` + Dir *string `json:"dir"` + Hook *string `json:"hook"` + Region *string `json:"region"` + KeyID *string `json:"key_id"` + ApplicationKey *string `json:"application_key"` + EndpointURL *string `json:"endpoint_url"` + Interval *string `json:"interval"` + HTTPTimeout *string `json:"http_timeout"` + Once *bool `json:"once"` + Prune *bool `json:"prune"` +} diff --git a/cmd/s3watch/main_test.go b/cmd/s3watch/main_test.go new file mode 100644 index 0000000..bd31000 --- /dev/null +++ b/cmd/s3watch/main_test.go @@ -0,0 +1,174 @@ +package main + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +func TestParseConfigLoadsJSONConfig(t *testing.T) { + configPath := writeConfig(t, `{ + "bucket": "example-bucket", + "prefix": "app/config", + "dir": "/srv/app/config", + "hook": "/usr/local/bin/reload-app", + "region": "us-east-1", + "key_id": "key-id-from-config", + "application_key": "application-key-from-config", + "endpoint_url": "https://s3.example.com", + "interval": "45s", + "http_timeout": "10s", + "once": true, + "prune": true +}`) + + cfg, err := parseConfig([]string{"-config", configPath}) + if err != nil { + t.Fatalf("parseConfig() error = %v", err) + } + + if cfg.bucket != "example-bucket" { + t.Fatalf("bucket = %q, want example-bucket", cfg.bucket) + } + if cfg.prefix != "app/config" { + t.Fatalf("prefix = %q, want app/config", cfg.prefix) + } + if cfg.dir != "/srv/app/config" { + t.Fatalf("dir = %q, want /srv/app/config", cfg.dir) + } + if cfg.hook != "/usr/local/bin/reload-app" { + t.Fatalf("hook = %q, want /usr/local/bin/reload-app", cfg.hook) + } + if cfg.region != "us-east-1" { + t.Fatalf("region = %q, want us-east-1", cfg.region) + } + if cfg.keyID != "key-id-from-config" { + t.Fatalf("keyID = %q, want key-id-from-config", cfg.keyID) + } + if cfg.applicationKey != "application-key-from-config" { + t.Fatalf("applicationKey = %q, want application-key-from-config", cfg.applicationKey) + } + if cfg.endpointURL != "https://s3.example.com" { + t.Fatalf("endpointURL = %q, want https://s3.example.com", cfg.endpointURL) + } + if cfg.interval != 45*time.Second { + t.Fatalf("interval = %s, want 45s", cfg.interval) + } + if cfg.httpTimeout != 10*time.Second { + t.Fatalf("httpTimeout = %s, want 10s", cfg.httpTimeout) + } + if !cfg.once { + t.Fatal("once = false, want true") + } + if !cfg.prune { + t.Fatal("prune = false, want true") + } +} + +func TestParseConfigAllowsFlagOverrides(t *testing.T) { + configPath := writeConfig(t, `{ + "bucket": "config-bucket", + "dir": "/from/config", + "interval": "1m", + "prune": true +}`) + + cfg, err := parseConfig([]string{ + "-config", configPath, + "-bucket", "flag-bucket", + "-dir", "/from/flag", + "-interval", "5s", + "-key-id", "flag-key-id", + "-application-key", "flag-application-key", + "-endpoint-url", "https://s3.flag.example.com", + "-prune=false", + }) + if err != nil { + t.Fatalf("parseConfig() error = %v", err) + } + + if cfg.bucket != "flag-bucket" { + t.Fatalf("bucket = %q, want flag-bucket", cfg.bucket) + } + if cfg.dir != "/from/flag" { + t.Fatalf("dir = %q, want /from/flag", cfg.dir) + } + if cfg.interval != 5*time.Second { + t.Fatalf("interval = %s, want 5s", cfg.interval) + } + if cfg.keyID != "flag-key-id" { + t.Fatalf("keyID = %q, want flag-key-id", cfg.keyID) + } + if cfg.applicationKey != "flag-application-key" { + t.Fatalf("applicationKey = %q, want flag-application-key", cfg.applicationKey) + } + if cfg.endpointURL != "https://s3.flag.example.com" { + t.Fatalf("endpointURL = %q, want https://s3.flag.example.com", cfg.endpointURL) + } + if cfg.prune { + t.Fatal("prune = true, want false") + } +} + +func TestConfigValidateRequiresCompleteStaticCredentials(t *testing.T) { + tests := []struct { + name string + cfg cliConfig + }{ + { + name: "missing application key", + cfg: cliConfig{bucket: "bucket", dir: "/tmp/sync", keyID: "key-id"}, + }, + { + name: "missing key id", + cfg: cliConfig{bucket: "bucket", dir: "/tmp/sync", applicationKey: "application-key"}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := defaultConfig() + cfg.bucket = tt.cfg.bucket + cfg.dir = tt.cfg.dir + cfg.keyID = tt.cfg.keyID + cfg.applicationKey = tt.cfg.applicationKey + if err := cfg.validate(); err == nil { + t.Fatal("validate() error = nil, want credential validation error") + } + }) + } +} + +func TestConfigValidateRejectsInvalidEndpointURL(t *testing.T) { + cfg := defaultConfig() + cfg.bucket = "bucket" + cfg.dir = "/tmp/sync" + cfg.endpointURL = "localhost:9000" + + if err := cfg.validate(); err == nil { + t.Fatal("validate() error = nil, want endpoint validation error") + } +} + +func TestParseConfigRejectsInvalidDuration(t *testing.T) { + configPath := writeConfig(t, `{ + "bucket": "example-bucket", + "dir": "/srv/app/config", + "interval": "soon" +}`) + + if _, err := parseConfig([]string{"-config", configPath}); err == nil { + t.Fatal("parseConfig() error = nil, want invalid duration error") + } +} + +func writeConfig(t *testing.T, content string) string { + t.Helper() + + path := filepath.Join(t.TempDir(), "s3watch.json") + if err := os.WriteFile(path, []byte(content), 0o644); err != nil { + t.Fatalf("writing config: %v", err) + } + return path +} diff --git a/examples/s3watch.json b/examples/s3watch.json new file mode 100644 index 0000000..8173afd --- /dev/null +++ b/examples/s3watch.json @@ -0,0 +1,14 @@ +{ + "bucket": "my-application-config", + "prefix": "production/web", + "dir": "/srv/myapp/config", + "hook": "/usr/local/bin/reload-myapp", + "region": "us-east-1", + "key_id": "REPLACE_WITH_KEY_ID", + "application_key": "REPLACE_WITH_APPLICATION_KEY", + "endpoint_url": "https://s3.us-east-1.amazonaws.com", + "interval": "1m", + "http_timeout": "30s", + "once": false, + "prune": true +} diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..05b1e69 --- /dev/null +++ b/go.mod @@ -0,0 +1,27 @@ +module s3watch + +go 1.24.4 + +require ( + github.com/aws/aws-sdk-go-v2 v1.42.1 + github.com/aws/aws-sdk-go-v2/config v1.32.30 + github.com/aws/aws-sdk-go-v2/credentials v1.19.29 + github.com/aws/aws-sdk-go-v2/service/s3 v1.105.2 +) + +require ( + github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.14 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.30 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.30 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.30 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.31 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.13 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.23 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.30 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.31 // indirect + github.com/aws/aws-sdk-go-v2/service/signin v1.4.1 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.32.1 // indirect + github.com/aws/aws-sdk-go-v2/service/ssooidc v1.37.1 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.44.1 // indirect + github.com/aws/smithy-go v1.27.3 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..8a806e7 --- /dev/null +++ b/go.sum @@ -0,0 +1,36 @@ +github.com/aws/aws-sdk-go-v2 v1.42.1 h1:9eOTgu1z/dVtYpNZ3/8/XbbaX0x/BqE3HUzAzs6K0ek= +github.com/aws/aws-sdk-go-v2 v1.42.1/go.mod h1:5pKeft2eJj+gElQ38Jqg4ibCqh+/AK33/0X3hip7IjM= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.14 h1:3IZY0XAJquT3aHzbkHfPzy4ACPcEjVG0x87KOwtpqGY= +github.com/aws/aws-sdk-go-v2/aws/protocol/eventstream v1.7.14/go.mod h1:zwM6veDkhGgQFqkBy+uT28AAYpLu+uFMlPl+rCg/73E= +github.com/aws/aws-sdk-go-v2/config v1.32.30 h1:XwsEzpTJfQYJbFicz/QMLwAZdyeNVVoOEkbF7R3gPJk= +github.com/aws/aws-sdk-go-v2/config v1.32.30/go.mod h1:Ud32SuMc+/9BGxfpSVld7HrE2o05JwKmXY4M3jOQNZU= +github.com/aws/aws-sdk-go-v2/credentials v1.19.29 h1:WHZGssHH887cO0ox07SIQZsFx3MKD4ps6w0xUEmnKYQ= +github.com/aws/aws-sdk-go-v2/credentials v1.19.29/go.mod h1:Mhl0xR6zjguiuj00XRx2wMx22sAltk7oya39sT7fdg8= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.30 h1:/hi1JADLEW9YYryEz1w4GQu0EtP23pP553Cf9KgsDV4= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.18.30/go.mod h1:/3AOgy4K17Dm4ucMZVC/MJkzy5kmfKUcINRHZyo0koQ= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.30 h1:xM/Is9cKMHa8Jj8zkvWhvrFkZsXJV9E+BB4g0HW0duQ= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.4.30/go.mod h1:WueJeNDZvK1fMYEWJIkcivBfEzUkTpBhzlrUKKY8EuA= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.30 h1:jn46zC9LdsVR/ZpMIJqMqb8hHv31BlLx3ulVqNspUOk= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.7.30/go.mod h1:1hTMsAgbdS/AtUi4bw8+gUuh1pceo+eXRLfpSuSQj3M= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.31 h1:3GUprIsfmGcC5SACIyB0e7E0BM1O1b3Erl5CePYIAeQ= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.4.31/go.mod h1:7PuV1yl5e2xnUbm+RqvVg5i2iBM8EyijZNoI9wsOoOc= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.13 h1:mbRIur/BiHK6SKPjoBIXSE/hJ6g6JGRLuxQy1jGjlN4= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.13/go.mod h1:ITg9em2KbJx1s0y4aqRX5OYWG6HBZ5TVR//OdpEZ2CQ= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.23 h1:9Fjh6fi/U5JEStVZijmaMpUwE/gvBJj7x2B/PjbO9To= +github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.23/go.mod h1:iMoT2f1tClxrWAAnKCXjZQ6LOmfLrMG14wmnWpM+F14= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.30 h1:/Z5jmNrKsSD7EmDjzAPsm/3L9IuOkzaynklJZ1qX7S4= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.30/go.mod h1:lEzEZnOosE7zi8Z6royW1cFJTD9fpab4Ul1SBrllewk= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.31 h1:uao4A3QZ5UmB326V6KF+qRpv9Tjz7IlnlnTbbANntlU= +github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.31/go.mod h1:I/1+z0VwL1GhQyLgkoHDlygpUZ+iTAwOQ/NsftiUL2I= +github.com/aws/aws-sdk-go-v2/service/s3 v1.105.2 h1:5C00eQYpTrgQXnp6V3P6P7zPElna3AXvlukbANE6nJI= +github.com/aws/aws-sdk-go-v2/service/s3 v1.105.2/go.mod h1:zdmCoFO/dSI7GlrwsPqFJI+WlFnSU4Tc8TJnlXrM1Do= +github.com/aws/aws-sdk-go-v2/service/signin v1.4.1 h1:V7ZZ300WPXGjvkyore5DGe0ljVPOxCXie/thWdtSBXE= +github.com/aws/aws-sdk-go-v2/service/signin v1.4.1/go.mod h1:mxC0nT/C8wMMS97DemZPzvUZxvIt+2Iq+eS3JdFZGgg= +github.com/aws/aws-sdk-go-v2/service/sso v1.32.1 h1:gYFYh4iLLcAOJRLNPY2aD2g9DIhKn4eof8UkIrr1rTk= +github.com/aws/aws-sdk-go-v2/service/sso v1.32.1/go.mod h1:u8af9Nqkmqnr96f7v9nHqzZT9XBwbXEkTiqT4ROuJSE= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.37.1 h1:arjT9Cm3/WYbGmD5TUZHk4UQn4Lle1fUNZs5FC6CtF0= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.37.1/go.mod h1:DMPWJBjYs6+3+f/qhBFEFPPlQ6NlhWjai3dJNvipJ84= +github.com/aws/aws-sdk-go-v2/service/sts v1.44.1 h1:RvfHDg+xvAeZ+5741vUEjpOVtYSIm93W2zhx10Xtydw= +github.com/aws/aws-sdk-go-v2/service/sts v1.44.1/go.mod h1:9gdl4RrflIdpDb2TlXshWgR1F9TeCkvqDx77Vpr4Z/Q= +github.com/aws/smithy-go v1.27.3 h1:F3Zb497UhhskkfpJmfkXswyo+t0sh9OTBnIHjogWbVY= +github.com/aws/smithy-go v1.27.3/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= diff --git a/internal/daemon/daemon.go b/internal/daemon/daemon.go new file mode 100644 index 0000000..68d60cc --- /dev/null +++ b/internal/daemon/daemon.go @@ -0,0 +1,94 @@ +// Package daemon coordinates periodic sync cycles and change hooks. +package daemon + +import ( + "context" + "fmt" + "log/slog" + "time" + + "s3watch/internal/syncer" +) + +// Config controls daemon scheduling behavior. +type Config struct { + Interval time.Duration + Once bool +} + +// Syncer synchronizes remote state into local state. +type Syncer interface { + Sync(context.Context) (syncer.Result, error) +} + +// HookRunner runs after a sync cycle applies changes. +type HookRunner interface { + Run(context.Context, syncer.Result) error +} + +// Run executes sync cycles until the context is canceled. +func Run(ctx context.Context, cfg Config, syncService Syncer, runner HookRunner, logger *slog.Logger) error { + if cfg.Interval <= 0 { + return fmt.Errorf("interval must be greater than zero") + } + if syncService == nil { + return fmt.Errorf("syncer is required") + } + if logger == nil { + logger = slog.Default() + } + + for { + if err := runCycle(ctx, syncService, runner, logger, cfg.Once); err != nil { + if cfg.Once { + return err + } + logger.Error("sync cycle failed", "error", err) + } + + if cfg.Once { + return nil + } + + timer := time.NewTimer(cfg.Interval) + select { + case <-ctx.Done(): + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + return nil + case <-timer.C: + } + } +} + +func runCycle(ctx context.Context, syncService Syncer, runner HookRunner, logger *slog.Logger, strict bool) error { + result, err := syncService.Sync(ctx) + if err != nil { + return fmt.Errorf("syncing: %w", err) + } + + logger.Info("sync cycle completed", + "downloaded", result.Downloaded, + "uploaded", result.Uploaded, + "updated", result.Updated, + "deleted", result.Deleted, + "unchanged", result.Unchanged, + ) + + if !result.Changed() || runner == nil { + return nil + } + + if err := runner.Run(ctx, result); err != nil { + if strict { + return fmt.Errorf("running hook: %w", err) + } + logger.Error("hook failed", "error", err) + } + + return nil +} diff --git a/internal/daemon/daemon_test.go b/internal/daemon/daemon_test.go new file mode 100644 index 0000000..138f5fe --- /dev/null +++ b/internal/daemon/daemon_test.go @@ -0,0 +1,76 @@ +package daemon + +import ( + "context" + "errors" + "log/slog" + "testing" + "time" + + "s3watch/internal/syncer" +) + +func TestRunOnceRunsHookOnlyWhenChanged(t *testing.T) { + tests := []struct { + name string + result syncer.Result + wantCalls int + }{ + { + name: "changed", + result: syncer.Result{Downloaded: 1}, + wantCalls: 1, + }, + { + name: "unchanged", + result: syncer.Result{Unchanged: 1}, + wantCalls: 0, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + hook := &fakeHook{} + err := Run(context.Background(), Config{Interval: time.Second, Once: true}, fakeSyncer{result: tt.result}, hook, slog.New(slog.DiscardHandler)) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if hook.calls != tt.wantCalls { + t.Fatalf("hook calls = %d, want %d", hook.calls, tt.wantCalls) + } + }) + } +} + +func TestRunOnceReturnsHookError(t *testing.T) { + wantErr := errors.New("hook failed") + err := Run( + context.Background(), + Config{Interval: time.Second, Once: true}, + fakeSyncer{result: syncer.Result{Updated: 1}}, + &fakeHook{err: wantErr}, + slog.New(slog.DiscardHandler), + ) + if !errors.Is(err, wantErr) { + t.Fatalf("Run() error = %v, want %v", err, wantErr) + } +} + +type fakeSyncer struct { + result syncer.Result + err error +} + +func (f fakeSyncer) Sync(context.Context) (syncer.Result, error) { + return f.result, f.err +} + +type fakeHook struct { + calls int + err error +} + +func (f *fakeHook) Run(context.Context, syncer.Result) error { + f.calls++ + return f.err +} diff --git a/internal/hook/hook.go b/internal/hook/hook.go new file mode 100644 index 0000000..e1447dc --- /dev/null +++ b/internal/hook/hook.go @@ -0,0 +1,48 @@ +// Package hook runs post-sync hook scripts. +package hook + +import ( + "context" + "fmt" + "os" + "os/exec" + "strconv" + + "s3watch/internal/syncer" +) + +// Runner executes a configured hook script after changes. +type Runner struct { + path string + dir string +} + +// NewRunner creates a hook runner for an executable script path. +func NewRunner(path string, dir string) Runner { + return Runner{path: path, dir: dir} +} + +// Run executes the hook and passes change counts through environment variables. +func (r Runner) Run(ctx context.Context, result syncer.Result) error { + if r.path == "" { + return nil + } + + cmd := exec.CommandContext(ctx, r.path) + cmd.Dir = r.dir + cmd.Stdout = os.Stdout + cmd.Stderr = os.Stderr + cmd.Env = append(os.Environ(), + "S3WATCH_CHANGED="+strconv.FormatBool(result.Changed()), + "S3WATCH_DOWNLOADED="+strconv.Itoa(result.Downloaded), + "S3WATCH_UPLOADED="+strconv.Itoa(result.Uploaded), + "S3WATCH_UPDATED="+strconv.Itoa(result.Updated), + "S3WATCH_DELETED="+strconv.Itoa(result.Deleted), + "S3WATCH_UNCHANGED="+strconv.Itoa(result.Unchanged), + ) + + if err := cmd.Run(); err != nil { + return fmt.Errorf("executing %q: %w", r.path, err) + } + return nil +} diff --git a/internal/hook/hook_test.go b/internal/hook/hook_test.go new file mode 100644 index 0000000..f3b27d5 --- /dev/null +++ b/internal/hook/hook_test.go @@ -0,0 +1,32 @@ +package hook + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + + "s3watch/internal/syncer" +) + +func TestRunnerRunPassesChangeEnvironment(t *testing.T) { + dir := t.TempDir() + script := filepath.Join(t.TempDir(), "hook.sh") + if err := os.WriteFile(script, []byte("#!/bin/sh\nprintf '%s,%s,%s,%s,%s,%s' \"$S3WATCH_CHANGED\" \"$S3WATCH_DOWNLOADED\" \"$S3WATCH_UPLOADED\" \"$S3WATCH_UPDATED\" \"$S3WATCH_DELETED\" \"$S3WATCH_UNCHANGED\" > hook.out\n"), 0o755); err != nil { + t.Fatalf("writing hook script: %v", err) + } + + result := syncer.Result{Downloaded: 2, Uploaded: 3, Updated: 4, Deleted: 5, Unchanged: 6} + if err := NewRunner(script, dir).Run(context.Background(), result); err != nil { + t.Fatalf("Run() error = %v", err) + } + + output, err := os.ReadFile(filepath.Join(dir, "hook.out")) + if err != nil { + t.Fatalf("reading hook output: %v", err) + } + if got, want := strings.TrimSpace(string(output)), "true,2,3,4,5,6"; got != want { + t.Fatalf("hook output = %q, want %q", got, want) + } +} diff --git a/internal/syncer/syncer.go b/internal/syncer/syncer.go new file mode 100644 index 0000000..4348164 --- /dev/null +++ b/internal/syncer/syncer.go @@ -0,0 +1,592 @@ +// 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) +} diff --git a/internal/syncer/syncer_test.go b/internal/syncer/syncer_test.go new file mode 100644 index 0000000..e3aa1ed --- /dev/null +++ b/internal/syncer/syncer_test.go @@ -0,0 +1,311 @@ +package syncer + +import ( + "context" + "io" + "os" + "path/filepath" + "strings" + "testing" + "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" +) + +func TestSyncDownloadsObjects(t *testing.T) { + modified := time.Unix(100, 0) + client := newFakeS3(map[string]fakeObject{ + "config/app.yml": {body: "port: 8080\n", modified: modified}, + }) + dir := t.TempDir() + + result, err := New(client, Config{Bucket: "bucket", Prefix: "config", Dir: dir}).Sync(context.Background()) + if err != nil { + t.Fatalf("Sync() error = %v", err) + } + if result != (Result{Downloaded: 1}) { + t.Fatalf("result = %+v, want one download", result) + } + + content, err := os.ReadFile(filepath.Join(dir, "app.yml")) + if err != nil { + t.Fatalf("reading downloaded file: %v", err) + } + if string(content) != "port: 8080\n" { + t.Fatalf("downloaded content = %q", content) + } + + info, err := os.Stat(filepath.Join(dir, "app.yml")) + if err != nil { + t.Fatalf("stat downloaded file: %v", err) + } + if info.ModTime().Unix() != modified.Unix() { + t.Fatalf("mod time = %s, want %s", info.ModTime(), modified) + } +} + +func TestSyncSkipsUnchangedObjects(t *testing.T) { + modified := time.Unix(200, 0) + client := newFakeS3(map[string]fakeObject{ + "app.yml": {body: "same", modified: modified}, + }) + dir := t.TempDir() + service := New(client, Config{Bucket: "bucket", Dir: dir}) + + if _, err := service.Sync(context.Background()); err != nil { + t.Fatalf("first Sync() error = %v", err) + } + client.getCalls = 0 + client.putCalls = 0 + + result, err := service.Sync(context.Background()) + if err != nil { + t.Fatalf("second Sync() error = %v", err) + } + if result != (Result{Unchanged: 1}) { + t.Fatalf("result = %+v, want one unchanged", result) + } + if client.getCalls != 0 { + t.Fatalf("GetObject calls = %d, want 0", client.getCalls) + } + if client.putCalls != 0 { + t.Fatalf("PutObject calls = %d, want 0", client.putCalls) + } +} + +func TestSyncDownloadsRemoteNewerObject(t *testing.T) { + firstModified := time.Unix(300, 0) + secondModified := time.Unix(301, 0) + client := newFakeS3(map[string]fakeObject{ + "app.yml": {body: "old", modified: firstModified}, + }) + dir := t.TempDir() + service := New(client, Config{Bucket: "bucket", Dir: dir}) + + if _, err := service.Sync(context.Background()); err != nil { + t.Fatalf("first Sync() error = %v", err) + } + + client.objects["app.yml"] = fakeObject{body: "new", modified: secondModified} + result, err := service.Sync(context.Background()) + if err != nil { + t.Fatalf("second Sync() error = %v", err) + } + if result != (Result{Updated: 1}) { + t.Fatalf("result = %+v, want one update", result) + } + + content, err := os.ReadFile(filepath.Join(dir, "app.yml")) + if err != nil { + t.Fatalf("reading updated file: %v", err) + } + if string(content) != "new" { + t.Fatalf("updated content = %q, want new", content) + } +} + +func TestSyncUploadsLocalOnlyFile(t *testing.T) { + client := newFakeS3(map[string]fakeObject{}) + dir := t.TempDir() + localPath := filepath.Join(dir, "local.txt") + writeFileAt(t, localPath, "local content", time.Unix(600, 0)) + + result, err := New(client, Config{Bucket: "bucket", Prefix: "config", Dir: dir}).Sync(context.Background()) + if err != nil { + t.Fatalf("Sync() error = %v", err) + } + if result != (Result{Uploaded: 1}) { + t.Fatalf("result = %+v, want one upload", result) + } + if got := client.objects["config/local.txt"].body; got != "local content" { + t.Fatalf("uploaded body = %q, want local content", got) + } +} + +func TestSyncUploadsLocalNewerObject(t *testing.T) { + client := newFakeS3(map[string]fakeObject{ + "app.yml": {body: "remote", modified: time.Unix(700, 0)}, + }) + dir := t.TempDir() + service := New(client, Config{Bucket: "bucket", Dir: dir}) + + if _, err := service.Sync(context.Background()); err != nil { + t.Fatalf("first Sync() error = %v", err) + } + localPath := filepath.Join(dir, "app.yml") + writeFileAt(t, localPath, "local", time.Unix(800, 0)) + + result, err := service.Sync(context.Background()) + if err != nil { + t.Fatalf("second Sync() error = %v", err) + } + if result != (Result{Updated: 1}) { + t.Fatalf("result = %+v, want one update", result) + } + if got := client.objects["app.yml"].body; got != "local" { + t.Fatalf("uploaded body = %q, want local", got) + } +} + +func TestSyncPruneDeletesUnchangedLocalAfterRemoteDelete(t *testing.T) { + client := newFakeS3(map[string]fakeObject{ + "stale.txt": {body: "remote", modified: time.Unix(400, 0)}, + }) + dir := t.TempDir() + service := New(client, Config{Bucket: "bucket", Dir: dir, Prune: true}) + + if _, err := service.Sync(context.Background()); err != nil { + t.Fatalf("first Sync() error = %v", err) + } + delete(client.objects, "stale.txt") + + result, err := service.Sync(context.Background()) + if err != nil { + t.Fatalf("second Sync() error = %v", err) + } + if result != (Result{Deleted: 1}) { + t.Fatalf("result = %+v, want one delete", result) + } + if _, err := os.Stat(filepath.Join(dir, "stale.txt")); !os.IsNotExist(err) { + t.Fatalf("stale file still exists or unexpected error: %v", err) + } +} + +func TestSyncPruneDeletesUnchangedRemoteAfterLocalDelete(t *testing.T) { + client := newFakeS3(map[string]fakeObject{ + "stale.txt": {body: "remote", modified: time.Unix(450, 0)}, + }) + dir := t.TempDir() + service := New(client, Config{Bucket: "bucket", Dir: dir, Prune: true}) + + if _, err := service.Sync(context.Background()); err != nil { + t.Fatalf("first Sync() error = %v", err) + } + if err := os.Remove(filepath.Join(dir, "stale.txt")); err != nil { + t.Fatalf("removing local file: %v", err) + } + + result, err := service.Sync(context.Background()) + if err != nil { + t.Fatalf("second Sync() error = %v", err) + } + if result != (Result{Deleted: 1}) { + t.Fatalf("result = %+v, want one delete", result) + } + if _, ok := client.objects["stale.txt"]; ok { + t.Fatal("remote object still exists") + } +} + +func TestSyncRejectsUnsafeKeys(t *testing.T) { + client := newFakeS3(map[string]fakeObject{ + "../secret.txt": {body: "secret", modified: time.Unix(500, 0)}, + }) + dir := t.TempDir() + + _, err := New(client, Config{Bucket: "bucket", Dir: dir}).Sync(context.Background()) + if err == nil { + t.Fatal("Sync() error = nil, want unsafe key error") + } + if !strings.Contains(err.Error(), "unsafe") { + t.Fatalf("Sync() error = %v, want unsafe key error", err) + } + if _, err := os.Stat(filepath.Join(filepath.Dir(dir), "secret.txt")); !os.IsNotExist(err) { + t.Fatalf("unsafe file was written or unexpected error: %v", err) + } +} + +type fakeObject struct { + body string + modified time.Time +} + +type fakeS3 struct { + objects map[string]fakeObject + getCalls int + putCalls int + delCalls int + now time.Time +} + +func newFakeS3(objects map[string]fakeObject) *fakeS3 { + return &fakeS3{ + objects: objects, + now: time.Unix(900, 0), + } +} + +func (f *fakeS3) ListObjectsV2(_ context.Context, input *s3.ListObjectsV2Input, _ ...func(*s3.Options)) (*s3.ListObjectsV2Output, error) { + var contents []types.Object + prefix := aws.ToString(input.Prefix) + for key, object := range f.objects { + if !strings.HasPrefix(key, prefix) { + continue + } + size := int64(len(object.body)) + modified := object.modified + contents = append(contents, types.Object{ + Key: aws.String(key), + LastModified: &modified, + Size: &size, + }) + } + truncated := false + return &s3.ListObjectsV2Output{ + Contents: contents, + IsTruncated: &truncated, + }, nil +} + +func (f *fakeS3) GetObject(_ context.Context, input *s3.GetObjectInput, _ ...func(*s3.Options)) (*s3.GetObjectOutput, error) { + f.getCalls++ + object := f.objects[aws.ToString(input.Key)] + return &s3.GetObjectOutput{ + Body: io.NopCloser(strings.NewReader(object.body)), + }, nil +} + +func (f *fakeS3) PutObject(_ context.Context, input *s3.PutObjectInput, _ ...func(*s3.Options)) (*s3.PutObjectOutput, error) { + f.putCalls++ + body, err := io.ReadAll(input.Body) + if err != nil { + return nil, err + } + f.objects[aws.ToString(input.Key)] = fakeObject{ + body: string(body), + modified: f.now, + } + return &s3.PutObjectOutput{}, nil +} + +func (f *fakeS3) HeadObject(_ context.Context, input *s3.HeadObjectInput, _ ...func(*s3.Options)) (*s3.HeadObjectOutput, error) { + object := f.objects[aws.ToString(input.Key)] + size := int64(len(object.body)) + modified := object.modified + return &s3.HeadObjectOutput{ + ContentLength: &size, + LastModified: &modified, + }, nil +} + +func (f *fakeS3) DeleteObject(_ context.Context, input *s3.DeleteObjectInput, _ ...func(*s3.Options)) (*s3.DeleteObjectOutput, error) { + f.delCalls++ + delete(f.objects, aws.ToString(input.Key)) + return &s3.DeleteObjectOutput{}, nil +} + +func writeFileAt(t *testing.T, path string, content string, modified time.Time) { + t.Helper() + + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatalf("creating parent dir: %v", err) + } + if err := os.WriteFile(path, []byte(content), 0o644); err != nil { + t.Fatalf("writing file: %v", err) + } + if err := os.Chtimes(path, modified, modified); err != nil { + t.Fatalf("setting mtime: %v", err) + } +}