diff --git a/README.md b/README.md index 58c306d..7be219e 100644 --- a/README.md +++ b/README.md @@ -116,6 +116,10 @@ ingest = "directory" # Local ingest directory (when ingest = "directory") ingest_dir = "~/paperless-ingest" +# Archive ingested files to prevent re-ingestion on Paperless restart (optional) +# ingest_archive_dir = "~/paperflow-archive" +# ingest_archive_after = "5m" + # Paperless API settings (when ingest = "api") # paperless_url = "https://paperless.example.com" # Token is stored separately in ~/.config/paperflow/token @@ -168,6 +172,8 @@ Flags override config values for a single run: | `--watch` | Override watch directory | | `--ingest` | Override ingestion method | | `--ingest-dir` | Override ingest directory | +| `--ingest-archive-dir` | Archive directory for ingested files | +| `--ingest-archive-after` | Delay before archiving (default: `5m`) | | `--paperless-url` | Paperless-ngx base URL (for API ingestion) | | `--paperless-token-file` | Path to file containing Paperless API token | | `--config` | Path to config file (default: `$XDG_CONFIG_HOME/paperflow/config.toml`) | @@ -184,6 +190,8 @@ Environment variables with the `PAPERFLOW_` prefix override config file values ( | `PAPERFLOW_INGEST` | Override ingestion method | | `PAPERFLOW_INGEST_DIR` | Override ingest directory | | `PAPERFLOW_PAPERLESS_URL` | Paperless-ngx base URL | +| `PAPERFLOW_INGEST_ARCHIVE_DIR` | Archive directory for ingested files | +| `PAPERFLOW_INGEST_ARCHIVE_AFTER` | Delay before archiving (e.g. `5m`) | | `PAPERFLOW_NO_NOTIFY` | Set to `1` or `true` to disable notifications | Config resolution order: defaults -> config file -> environment variables -> CLI flags. @@ -225,6 +233,21 @@ Checks the config file for errors and verifies that: - Paperless URL is reachable (if using API ingestion) - Token file exists and has correct permissions +## Ingest archive (directory mode) + +When using directory-mode ingestion, Paperless-ngx may re-ingest files from its consume directory after a restart. To prevent this, paperflow can automatically move files from the consume directory to a separate archive after a configurable delay: + +```toml +ingest_archive_dir = "~/paperflow-archive" +ingest_archive_after = "5m" +``` + +After copying a file to the consume directory, paperflow waits for the configured delay (giving Paperless time to pick it up), then moves the file to the archive directory with a timestamp prefix (e.g. `20260421-164532_invoice.pdf`). If Paperless already consumed and deleted the file, the archiver silently skips it. + +This is disabled by default. The archive also serves as an audit trail of what was sent to Paperless — files can be manually re-ingested from the archive if needed. + +This does not apply to API mode, which uploads directly and has no residual files. + ## Ingestible file types By default, the following file types are forwarded to Paperless when ingestion is enabled: diff --git a/VERSION b/VERSION index 9e11b32..1d0ba9e 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.3.1 +0.4.0 diff --git a/cmd/paperflow/init_cmd.go b/cmd/paperflow/init_cmd.go index 3c116d4..8216a29 100644 --- a/cmd/paperflow/init_cmd.go +++ b/cmd/paperflow/init_cmd.go @@ -52,7 +52,7 @@ func runInit(f flags) error { return fmt.Errorf("invalid ingestion method: %s", ingest) } - var ingestDir, paperlessURL, token string + var ingestDir, archiveDir, archiveAfter, paperlessURL, token string switch ingest { case "directory": @@ -63,6 +63,24 @@ func runInit(f flags) error { ingestDir = "~/paperless-ingest" } + fmt.Print("Archive ingested files to prevent re-ingestion on Paperless restart? [y/N] ") + archiveAnswer, _ := reader.ReadString('\n') + archiveAnswer = strings.TrimSpace(strings.ToLower(archiveAnswer)) + if archiveAnswer == "y" || archiveAnswer == "yes" { + fmt.Print("Archive directory [~/paperflow-archive]: ") + archiveDir, _ = reader.ReadString('\n') + archiveDir = strings.TrimSpace(archiveDir) + if archiveDir == "" { + archiveDir = "~/paperflow-archive" + } + fmt.Print("Archive delay [5m]: ") + archiveAfter, _ = reader.ReadString('\n') + archiveAfter = strings.TrimSpace(archiveAfter) + if archiveAfter == "" { + archiveAfter = "5m" + } + } + case "api": fmt.Print("Paperless URL (e.g. https://paperless.example.com): ") paperlessURL, _ = reader.ReadString('\n') @@ -87,6 +105,10 @@ func runInit(f flags) error { if ingest == "directory" { fmt.Fprintf(&b, "ingest_dir = %q\n", ingestDir) + if archiveDir != "" { + fmt.Fprintf(&b, "ingest_archive_dir = %q\n", archiveDir) + fmt.Fprintf(&b, "ingest_archive_after = %q\n", archiveAfter) + } } if ingest == "api" { fmt.Fprintf(&b, "paperless_url = %q\n", paperlessURL) diff --git a/cmd/paperflow/main.go b/cmd/paperflow/main.go index 0e333ed..6124dd7 100644 --- a/cmd/paperflow/main.go +++ b/cmd/paperflow/main.go @@ -17,6 +17,8 @@ type flags struct { watchDir string ingest string ingestDir string + ingestArchiveDir string + ingestArchiveAfter string paperlessURL string paperlessTokenFile string config string @@ -83,6 +85,8 @@ Flags: --watch Override watch directory --ingest Override ingestion method (directory, api, none) --ingest-dir Override ingest directory + --ingest-archive-dir Archive directory for ingested files + --ingest-archive-after Delay before archiving (default: 5m) --paperless-url Paperless-ngx base URL (for API ingestion) --paperless-token-file

Path to Paperless API token file --config Path to config file @@ -101,7 +105,7 @@ func findCommand(args []string) string { if strings.HasPrefix(arg, "--") { // Flags that take a value. switch arg { - case "--watch", "--ingest", "--ingest-dir", "--paperless-url", "--paperless-token-file", "--config": + case "--watch", "--ingest", "--ingest-dir", "--ingest-archive-dir", "--ingest-archive-after", "--paperless-url", "--paperless-token-file", "--config": skip = true } continue @@ -141,6 +145,16 @@ func parseFlags(args []string) flags { i++ f.ingestDir = args[i] } + case "--ingest-archive-dir": + if i+1 < len(args) { + i++ + f.ingestArchiveDir = args[i] + } + case "--ingest-archive-after": + if i+1 < len(args) { + i++ + f.ingestArchiveAfter = args[i] + } case "--paperless-url": if i+1 < len(args) { i++ @@ -210,6 +224,12 @@ func loadConfigWithFlags(f flags) (*config.Config, error) { } cfg.Token = strings.TrimSpace(string(data)) } + if f.ingestArchiveDir != "" { + cfg.IngestArchiveDir = f.ingestArchiveDir + } + if f.ingestArchiveAfter != "" { + cfg.IngestArchiveAfter = f.ingestArchiveAfter + } if f.noNotify { cfg.Notifications.Enabled = false } diff --git a/cmd/paperflow/service_cmd.go b/cmd/paperflow/service_cmd.go index 1a4ad13..ee63846 100644 --- a/cmd/paperflow/service_cmd.go +++ b/cmd/paperflow/service_cmd.go @@ -47,6 +47,12 @@ func serviceInstall(f flags) error { if f.ingestDir != "" { extraFlags = append(extraFlags, "--ingest-dir", f.ingestDir) } + if f.ingestArchiveDir != "" { + extraFlags = append(extraFlags, "--ingest-archive-dir", f.ingestArchiveDir) + } + if f.ingestArchiveAfter != "" { + extraFlags = append(extraFlags, "--ingest-archive-after", f.ingestArchiveAfter) + } if f.paperlessURL != "" { extraFlags = append(extraFlags, "--paperless-url", f.paperlessURL) } diff --git a/cmd/paperflow/validate_cmd.go b/cmd/paperflow/validate_cmd.go index a033b82..d2b57f8 100644 --- a/cmd/paperflow/validate_cmd.go +++ b/cmd/paperflow/validate_cmd.go @@ -3,6 +3,7 @@ package main import ( "fmt" "os" + "time" "github.com/alcxyz/paperflow/internal/config" ) @@ -53,6 +54,25 @@ func runValidate(f flags) error { } else { fmt.Printf(" OK ingest_dir: %s\n", ingestDir) } + + if cfg.IngestArchiveDir != "" { + archiveDir := config.ExpandTilde(cfg.IngestArchiveDir) + if info, err := os.Stat(archiveDir); err != nil { + fmt.Printf(" WARN ingest_archive_dir: %s does not exist (will be created)\n", archiveDir) + warnings++ + } else if !info.IsDir() { + fmt.Printf(" FAIL ingest_archive_dir: %s is not a directory\n", archiveDir) + errors++ + } else { + fmt.Printf(" OK ingest_archive_dir: %s\n", archiveDir) + } + if _, err := time.ParseDuration(cfg.IngestArchiveAfter); err != nil { + fmt.Printf(" FAIL ingest_archive_after: invalid duration %q\n", cfg.IngestArchiveAfter) + errors++ + } else { + fmt.Printf(" OK ingest_archive_after: %s\n", cfg.IngestArchiveAfter) + } + } case "api": if cfg.PaperlessURL == "" { fmt.Println(" FAIL paperless_url: not set") diff --git a/docs/adr/ADR-014-ingest-archive.md b/docs/adr/ADR-014-ingest-archive.md new file mode 100644 index 0000000..596a274 --- /dev/null +++ b/docs/adr/ADR-014-ingest-archive.md @@ -0,0 +1,35 @@ +# ADR-014: Archive ingested files from consume directory + +**Status:** Accepted +**Date:** 2026-04-21 +**Applies to:** `internal/ingest/archiver.go`, `internal/config/config.go` + +## Context + +When using directory-mode ingestion, paperflow copies files into Paperless-ngx's consume directory. If Paperless restarts (e.g. container redeployment, auto-update), it re-scans the consume directory and re-ingests any files still present, potentially creating duplicates. API mode does not have this problem since uploads are fire-and-forget. + +## Decision + +Add an optional archiver that moves files from the consume directory to a separate archive directory after a configurable delay. The delay gives Paperless time to pick up the file first. + +Configuration: +- `ingest_archive_dir`: path to the archive directory (empty = disabled, the default) +- `ingest_archive_after`: delay before archiving (default: `"5m"`) + +Archive filenames use a flat layout with timestamp prefix: `20260421-164532_invoice.pdf`. The original organized copy in the watch directory is untouched. + +If the file is already gone when the timer fires (Paperless consumed and deleted it), the archiver silently skips. On shutdown, all pending archives are flushed immediately. + +## Alternatives Considered + +- **Delete from consume dir after delay:** Simpler but loses the safety net. If Paperless missed a file, it's gone. +- **Document only, no mitigation:** Honest but doesn't solve the problem for users who can't switch to API mode. +- **Use API mode instead:** Sidesteps the problem entirely but requires network access to Paperless. Some deployments use local directory mounts by design. + +## Consequences + +- Directory-mode users can opt in to archive and avoid re-ingestion on Paperless restart. +- The archive directory serves as an audit trail of what was sent to Paperless. +- Files can be manually re-ingested from the archive if needed. +- Disabled by default — zero behavior change for existing users. +- Adds a per-file timer goroutine; memory is bounded by the number of files ingested within the delay window. diff --git a/docs/adr/README.md b/docs/adr/README.md index 801f605..69d51e0 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -15,3 +15,4 @@ | [ADR-011](ADR-011-git-derived-version.md) | ~~Version derived from git, no manual bumping~~ (superseded by ADR-012) | build | | [ADR-012](ADR-012-version-file-auto-tag.md) | VERSION file with CI auto-tagging | build, CI | | [ADR-013](ADR-013-systemd-path-injection.md) | Inject PATH into generated systemd unit | service | +| [ADR-014](ADR-014-ingest-archive.md) | Archive ingested files from consume directory | ingest | diff --git a/internal/config/config.go b/internal/config/config.go index 8f8f345..dfa9301 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -15,7 +15,9 @@ type Config struct { Ingest string `toml:"ingest"` IngestDir string `toml:"ingest_dir"` - PaperlessURL string `toml:"paperless_url"` + PaperlessURL string `toml:"paperless_url"` + IngestArchiveDir string `toml:"ingest_archive_dir"` + IngestArchiveAfter string `toml:"ingest_archive_after"` Notifications NotificationsConfig `toml:"notifications"` Buckets map[string][]string `toml:"buckets"` @@ -51,7 +53,8 @@ func DefaultConfig() *Config { return &Config{ WatchDir: "~/Documents", Ingest: "none", - IngestDir: "~/paperless-ingest", + IngestDir: "~/paperless-ingest", + IngestArchiveAfter: "5m", Notifications: NotificationsConfig{ Enabled: true, BatchWindow: "3s", @@ -130,6 +133,9 @@ func LoadConfig(path string) (*Config, error) { if cfg.IngestDir == "" { cfg.IngestDir = defaults.IngestDir } + if cfg.IngestArchiveAfter == "" { + cfg.IngestArchiveAfter = defaults.IngestArchiveAfter + } if cfg.Notifications.BatchWindow == "" { cfg.Notifications = defaults.Notifications } @@ -146,6 +152,7 @@ func LoadConfig(path string) (*Config, error) { // Expand tildes in paths. cfg.WatchDir = ExpandTilde(cfg.WatchDir) cfg.IngestDir = ExpandTilde(cfg.IngestDir) + cfg.IngestArchiveDir = ExpandTilde(cfg.IngestArchiveDir) // Apply environment variable overrides. applyEnvOverrides(&cfg) @@ -199,6 +206,12 @@ func applyEnvOverrides(cfg *Config) { if v := os.Getenv("PAPERFLOW_PAPERLESS_URL"); v != "" { cfg.PaperlessURL = v } + if v := os.Getenv("PAPERFLOW_INGEST_ARCHIVE_DIR"); v != "" { + cfg.IngestArchiveDir = ExpandTilde(v) + } + if v := os.Getenv("PAPERFLOW_INGEST_ARCHIVE_AFTER"); v != "" { + cfg.IngestArchiveAfter = v + } if v := os.Getenv("PAPERFLOW_NO_NOTIFY"); v == "1" || v == "true" { cfg.Notifications.Enabled = false } diff --git a/internal/ingest/archiver.go b/internal/ingest/archiver.go new file mode 100644 index 0000000..18b51c8 --- /dev/null +++ b/internal/ingest/archiver.go @@ -0,0 +1,125 @@ +package ingest + +import ( + "fmt" + "io" + "log" + "os" + "path/filepath" + "sync" + "time" +) + +// Archiver moves files from the ingest directory to an archive directory +// after a configurable delay. This prevents Paperless-ngx from re-ingesting +// files if it restarts and re-scans its consume directory. +type Archiver struct { + archiveDir string + delay time.Duration + + mu sync.Mutex + pending map[string]*time.Timer +} + +// NewArchiver creates an Archiver. If archiveDir is empty, returns nil +// (feature disabled). The caller should nil-check before calling methods. +func NewArchiver(archiveDir string, delayStr string) (*Archiver, error) { + if archiveDir == "" { + return nil, nil + } + delay, err := time.ParseDuration(delayStr) + if err != nil { + return nil, fmt.Errorf("parsing ingest_archive_after %q: %w", delayStr, err) + } + return &Archiver{ + archiveDir: archiveDir, + delay: delay, + pending: make(map[string]*time.Timer), + }, nil +} + +// Schedule queues a file for archival after the configured delay. +func (a *Archiver) Schedule(ingestPath string) { + a.mu.Lock() + defer a.mu.Unlock() + + timer := time.AfterFunc(a.delay, func() { + a.archiveFile(ingestPath) + }) + a.pending[ingestPath] = timer + log.Printf("archive scheduled for %s in %s", filepath.Base(ingestPath), a.delay) +} + +// Close flushes all pending archives immediately (for graceful shutdown). +func (a *Archiver) Close() { + a.mu.Lock() + pending := make(map[string]*time.Timer, len(a.pending)) + for k, v := range a.pending { + pending[k] = v + } + a.mu.Unlock() + + for path, timer := range pending { + timer.Stop() + a.archiveFile(path) + } +} + +// archiveFile moves a single file from the ingest dir to the archive dir. +func (a *Archiver) archiveFile(ingestPath string) { + a.mu.Lock() + delete(a.pending, ingestPath) + a.mu.Unlock() + + filename := filepath.Base(ingestPath) + + // File may already be consumed by Paperless. + if _, err := os.Stat(ingestPath); os.IsNotExist(err) { + log.Printf("archive: %s already consumed, skipping", filename) + return + } + + if err := os.MkdirAll(a.archiveDir, 0755); err != nil { + log.Printf("archive: failed to create dir %s: %v", a.archiveDir, err) + return + } + + ts := time.Now().Format("20060102-150405") + archiveName := fmt.Sprintf("%s_%s", ts, filename) + destPath := filepath.Join(a.archiveDir, archiveName) + destPath = ResolveCollision(destPath) + + if err := moveFile(ingestPath, destPath); err != nil { + log.Printf("archive: failed to move %s: %v", filename, err) + return + } + + log.Printf("archived %s -> %s", filename, destPath) +} + +// moveFile moves src to dst, falling back to copy+remove for cross-device moves. +func moveFile(src, dst string) error { + if err := os.Rename(src, dst); err == nil { + return nil + } + + in, err := os.Open(src) + if err != nil { + return err + } + defer in.Close() + + out, err := os.Create(dst) + if err != nil { + return err + } + defer out.Close() + + if _, err := io.Copy(out, in); err != nil { + return err + } + if err := out.Close(); err != nil { + return err + } + return os.Remove(src) +} diff --git a/internal/ingest/archiver_test.go b/internal/ingest/archiver_test.go new file mode 100644 index 0000000..e9c7bdf --- /dev/null +++ b/internal/ingest/archiver_test.go @@ -0,0 +1,171 @@ +package ingest + +import ( + "os" + "path/filepath" + "strings" + "testing" + "time" +) + +func TestNewArchiver_DisabledWhenEmpty(t *testing.T) { + a, err := NewArchiver("", "5m") + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if a != nil { + t.Error("expected nil archiver when dir is empty") + } +} + +func TestNewArchiver_InvalidDuration(t *testing.T) { + _, err := NewArchiver("/tmp/archive", "bad") + if err == nil { + t.Error("expected error for invalid duration") + } +} + +func TestArchiver_ScheduleAndWait(t *testing.T) { + tmp := t.TempDir() + ingestDir := filepath.Join(tmp, "ingest") + archiveDir := filepath.Join(tmp, "archive") + if err := os.MkdirAll(ingestDir, 0755); err != nil { + t.Fatal(err) + } + + // Create a file in the ingest dir. + ingestFile := filepath.Join(ingestDir, "invoice.pdf") + if err := os.WriteFile(ingestFile, []byte("pdf content"), 0644); err != nil { + t.Fatal(err) + } + + a, err := NewArchiver(archiveDir, "50ms") + if err != nil { + t.Fatal(err) + } + + a.Schedule(ingestFile) + + // Wait for the timer to fire. + time.Sleep(150 * time.Millisecond) + + // File should be gone from ingest dir. + if _, err := os.Stat(ingestFile); !os.IsNotExist(err) { + t.Error("file should have been moved from ingest dir") + } + + // File should be in archive dir with timestamp prefix. + entries, err := os.ReadDir(archiveDir) + if err != nil { + t.Fatalf("ReadDir: %v", err) + } + if len(entries) != 1 { + t.Fatalf("expected 1 archived file, got %d", len(entries)) + } + name := entries[0].Name() + if !strings.HasSuffix(name, "_invoice.pdf") { + t.Errorf("expected timestamp_invoice.pdf, got %q", name) + } + + // Verify content. + data, err := os.ReadFile(filepath.Join(archiveDir, name)) + if err != nil { + t.Fatal(err) + } + if string(data) != "pdf content" { + t.Errorf("content = %q, want %q", string(data), "pdf content") + } +} + +func TestArchiver_CloseFlushesImmediately(t *testing.T) { + tmp := t.TempDir() + ingestDir := filepath.Join(tmp, "ingest") + archiveDir := filepath.Join(tmp, "archive") + if err := os.MkdirAll(ingestDir, 0755); err != nil { + t.Fatal(err) + } + + ingestFile := filepath.Join(ingestDir, "invoice.pdf") + if err := os.WriteFile(ingestFile, []byte("data"), 0644); err != nil { + t.Fatal(err) + } + + a, err := NewArchiver(archiveDir, "1h") // Long delay. + if err != nil { + t.Fatal(err) + } + + a.Schedule(ingestFile) + a.Close() + + // File should be archived immediately. + if _, err := os.Stat(ingestFile); !os.IsNotExist(err) { + t.Error("file should have been archived on Close") + } + + entries, err := os.ReadDir(archiveDir) + if err != nil { + t.Fatal(err) + } + if len(entries) != 1 { + t.Fatalf("expected 1 archived file, got %d", len(entries)) + } +} + +func TestArchiver_MissingFileSkipped(t *testing.T) { + tmp := t.TempDir() + archiveDir := filepath.Join(tmp, "archive") + + a, err := NewArchiver(archiveDir, "50ms") + if err != nil { + t.Fatal(err) + } + + // Schedule a file that doesn't exist. + a.Schedule(filepath.Join(tmp, "nonexistent.pdf")) + + time.Sleep(150 * time.Millisecond) + + // Archive dir should not have been created (nothing to archive). + if _, err := os.Stat(archiveDir); !os.IsNotExist(err) { + entries, _ := os.ReadDir(archiveDir) + if len(entries) > 0 { + t.Error("no files should be archived for missing source") + } + } +} + +func TestArchiver_MultipleFiles(t *testing.T) { + tmp := t.TempDir() + ingestDir := filepath.Join(tmp, "ingest") + archiveDir := filepath.Join(tmp, "archive") + if err := os.MkdirAll(ingestDir, 0755); err != nil { + t.Fatal(err) + } + + files := []string{"a.pdf", "b.pdf", "c.pdf"} + for _, f := range files { + if err := os.WriteFile(filepath.Join(ingestDir, f), []byte(f), 0644); err != nil { + t.Fatal(err) + } + } + + a, err := NewArchiver(archiveDir, "1h") + if err != nil { + t.Fatal(err) + } + + for _, f := range files { + a.Schedule(filepath.Join(ingestDir, f)) + } + + a.Close() + + entries, err := os.ReadDir(archiveDir) + if err != nil { + t.Fatal(err) + } + if len(entries) != 3 { + t.Fatalf("expected 3 archived files, got %d", len(entries)) + } +} diff --git a/internal/ingest/directory.go b/internal/ingest/directory.go index 67b26c9..549f2f0 100644 --- a/internal/ingest/directory.go +++ b/internal/ingest/directory.go @@ -10,7 +10,8 @@ import ( ) // IngestDirectory copies the file to the configured ingest directory. -func IngestDirectory(path string, ingestDir string) error { +// It returns the destination path on success. +func IngestDirectory(path string, ingestDir string) (string, error) { filename := filepath.Base(path) destPath := filepath.Join(ingestDir, filename) @@ -18,27 +19,27 @@ func IngestDirectory(path string, ingestDir string) error { destPath = ResolveCollision(destPath) if err := os.MkdirAll(ingestDir, 0755); err != nil { - return fmt.Errorf("creating ingest directory: %w", err) + return "", fmt.Errorf("creating ingest directory: %w", err) } in, err := os.Open(path) if err != nil { - return err + return "", err } defer in.Close() out, err := os.Create(destPath) if err != nil { - return err + return "", err } defer out.Close() if _, err := io.Copy(out, in); err != nil { - return err + return "", err } log.Printf("ingested %s -> %s", filename, destPath) - return out.Close() + return destPath, out.Close() } // ResolveCollision returns a unique destination path. If destPath already diff --git a/internal/ingest/directory_test.go b/internal/ingest/directory_test.go index e8fe4fe..b68aa7d 100644 --- a/internal/ingest/directory_test.go +++ b/internal/ingest/directory_test.go @@ -20,12 +20,16 @@ func TestIngestDirectory(t *testing.T) { t.Fatal(err) } - if err := IngestDirectory(srcFile, ingestDir); err != nil { + destPath, err := IngestDirectory(srcFile, ingestDir) + if err != nil { t.Fatalf("IngestDirectory: %v", err) } // File should exist in ingest dir. destFile := filepath.Join(ingestDir, "invoice.pdf") + if destPath != destFile { + t.Errorf("returned path = %q, want %q", destPath, destFile) + } data, err := os.ReadFile(destFile) if err != nil { t.Fatalf("ingested file not found: %v", err) @@ -62,7 +66,7 @@ func TestIngestDirectory_Collision(t *testing.T) { t.Fatal(err) } - if err := IngestDirectory(srcFile, ingestDir); err != nil { + if _, err := IngestDirectory(srcFile, ingestDir); err != nil { t.Fatalf("IngestDirectory: %v", err) } @@ -84,7 +88,7 @@ func TestIngestDirectory_CreatesDir(t *testing.T) { } ingestDir := filepath.Join(tmp, "new", "nested", "ingest") - if err := IngestDirectory(srcFile, ingestDir); err != nil { + if _, err := IngestDirectory(srcFile, ingestDir); err != nil { t.Fatalf("IngestDirectory: %v", err) } diff --git a/internal/organizer/organizer.go b/internal/organizer/organizer.go index a900f11..d994ac1 100644 --- a/internal/organizer/organizer.go +++ b/internal/organizer/organizer.go @@ -24,12 +24,14 @@ type Result struct { // Organizer handles sorting files into bucket/year/month directories. type Organizer struct { - config *config.Config + config *config.Config + archiver *ingest.Archiver } // NewOrganizer creates an Organizer with the given config. -func NewOrganizer(cfg *config.Config) *Organizer { - return &Organizer{config: cfg} +// The archiver may be nil if archive is disabled. +func NewOrganizer(cfg *config.Config, archiver *ingest.Archiver) *Organizer { + return &Organizer{config: cfg, archiver: archiver} } // ProcessFile sorts a file into the appropriate bucket/year/month directory @@ -129,7 +131,14 @@ func moveFile(src, dst string) error { func (o *Organizer) doIngest(path string) error { switch o.config.Ingest { case "directory": - return ingest.IngestDirectory(path, o.config.IngestDir) + destPath, err := ingest.IngestDirectory(path, o.config.IngestDir) + if err != nil { + return err + } + if o.archiver != nil { + o.archiver.Schedule(destPath) + } + return nil case "api": return ingest.IngestAPI(path, o.config.PaperlessURL, o.config.Token) default: diff --git a/internal/organizer/organizer_test.go b/internal/organizer/organizer_test.go index dd1873c..9805085 100644 --- a/internal/organizer/organizer_test.go +++ b/internal/organizer/organizer_test.go @@ -27,7 +27,7 @@ func testConfig(watchDir string) *config.Config { func TestProcessFile_BucketSorting(t *testing.T) { tmp := t.TempDir() cfg := testConfig(tmp) - org := NewOrganizer(cfg) + org := NewOrganizer(cfg, nil) // Create a test PDF file. src := filepath.Join(tmp, "invoice.pdf") @@ -62,7 +62,7 @@ func TestProcessFile_BucketSorting(t *testing.T) { func TestProcessFile_MiscBucket(t *testing.T) { tmp := t.TempDir() cfg := testConfig(tmp) - org := NewOrganizer(cfg) + org := NewOrganizer(cfg, nil) src := filepath.Join(tmp, "readme.txt") if err := os.WriteFile(src, []byte("hello"), 0644); err != nil { @@ -82,7 +82,7 @@ func TestProcessFile_MiscBucket(t *testing.T) { func TestProcessFile_ImageBucket(t *testing.T) { tmp := t.TempDir() cfg := testConfig(tmp) - org := NewOrganizer(cfg) + org := NewOrganizer(cfg, nil) src := filepath.Join(tmp, "photo.jpg") if err := os.WriteFile(src, []byte("fake jpg"), 0644); err != nil { @@ -102,7 +102,7 @@ func TestProcessFile_ImageBucket(t *testing.T) { func TestProcessFile_CollisionHandling(t *testing.T) { tmp := t.TempDir() cfg := testConfig(tmp) - org := NewOrganizer(cfg) + org := NewOrganizer(cfg, nil) // Create the first file and process it. src1 := filepath.Join(tmp, "invoice.pdf") @@ -167,7 +167,7 @@ func TestProcessFile_DryRun(t *testing.T) { tmp := t.TempDir() cfg := testConfig(tmp) cfg.DryRun = true - org := NewOrganizer(cfg) + org := NewOrganizer(cfg, nil) src := filepath.Join(tmp, "invoice.pdf") if err := os.WriteFile(src, []byte("fake pdf"), 0644); err != nil { @@ -200,7 +200,7 @@ func TestProcessFile_DirectoryIngestion(t *testing.T) { cfg := testConfig(tmp) cfg.Ingest = "directory" cfg.IngestDir = filepath.Join(tmp, "ingest") - org := NewOrganizer(cfg) + org := NewOrganizer(cfg, nil) src := filepath.Join(tmp, "invoice.pdf") if err := os.WriteFile(src, []byte("fake pdf"), 0644); err != nil { @@ -228,7 +228,7 @@ func TestProcessFile_MiscNotIngested(t *testing.T) { cfg := testConfig(tmp) cfg.Ingest = "directory" cfg.IngestDir = filepath.Join(tmp, "ingest") - org := NewOrganizer(cfg) + org := NewOrganizer(cfg, nil) src := filepath.Join(tmp, "readme.txt") if err := os.WriteFile(src, []byte("hello"), 0644); err != nil { diff --git a/internal/watcher/watcher.go b/internal/watcher/watcher.go index 1ba7f94..48614f7 100644 --- a/internal/watcher/watcher.go +++ b/internal/watcher/watcher.go @@ -15,6 +15,7 @@ import ( "github.com/alcxyz/paperflow/internal/ingest" "github.com/alcxyz/paperflow/internal/notify" "github.com/alcxyz/paperflow/internal/organizer" + "github.com/fsnotify/fsnotify" ) @@ -28,6 +29,7 @@ type Watcher struct { config *config.Config organizer *organizer.Organizer notifier *notify.Notifier + archiver *ingest.Archiver mu sync.Mutex seen map[string]time.Time @@ -35,10 +37,15 @@ type Watcher struct { // NewWatcher creates a Watcher with the given config. func NewWatcher(cfg *config.Config) (*Watcher, error) { + archiver, err := ingest.NewArchiver(cfg.IngestArchiveDir, cfg.IngestArchiveAfter) + if err != nil { + return nil, fmt.Errorf("creating archiver: %w", err) + } return &Watcher{ config: cfg, - organizer: organizer.NewOrganizer(cfg), + organizer: organizer.NewOrganizer(cfg, archiver), notifier: notify.NewNotifier(cfg), + archiver: archiver, seen: make(map[string]time.Time), }, nil } @@ -111,6 +118,9 @@ func (w *Watcher) Run() error { case sig := <-sigCh: log.Printf("received %s, shutting down", sig) + if w.archiver != nil { + w.archiver.Close() + } w.notifier.Close() return nil }