feat: add local cache

7 files changed, 112 insertions, 12 deletions

diff --git a/.gitignore b/.gitignore
index 67ec59b..186750e 100644
--- a/.gitignore
+++ b/.gitignore
@@ -34,3 +34,5 @@ go.work.sum
 
 config*.json
 !config*.sample.json
+
+cache.csv
diff --git a/config/config-b2.sample.json b/config/config-b2.sample.json
index 043c1a6..372972e 100644
--- a/config/config-b2.sample.json
+++ b/config/config-b2.sample.json
@@ -18,7 +18,9 @@
             "jpg",
             "jpeg",
             "png"
-        ]
+        ],
+        "CacheProcessed": true,
+        "CacheProcessedCsvPath": "cache.csv"
     },
     "Converters": [
         {
diff --git a/config/config.go b/config/config.go
index ab051b2..1620d75 100644
--- a/config/config.go
+++ b/config/config.go
@@ -19,8 +19,10 @@ type Config struct {
 }
 
 type InputConfig struct {
-	Storage         InputStorageConfig `json:"Storage" validate:"required"`
-	KnownExtensions []string           `json:"KnownExtensions" validate:"required,min=0,dive,min=1"`
+	Storage               InputStorageConfig `json:"Storage" validate:"required"`
+	KnownExtensions       []string           `json:"KnownExtensions" validate:"required,min=0,dive,min=1"`
+	CacheProcessed        bool               `json:"CacheProcessed"`
+	CacheProcessedCsvPath string             `json:"CacheProcessedCsvPath" validate:"filepath"`
 }
 
 type InputStorageConfig struct {
@@ -159,7 +161,7 @@ type OutputConfig struct {
 }
 
 type OutputLocalUnixConfig struct {
-	Path                     string `json:"Path" validate:"required,min=1"`
+	Path                     string `json:"Path" validate:"required,min=1,dirpath"`
 	DirPermissionMode        string `json:"DirPermissionMode" validate:"required,min=3"`
 	FilePermissionMode       string `json:"FilePermissionMode" validate:"required,min=3"`
 	AttributesImplementation string `json:"AttributesImplementation" validate:"required,oneof=xattr none"`
diff --git a/internal/client/input/b2.go b/internal/client/input/b2.go
index db68268..fe94333 100644
--- a/internal/client/input/b2.go
+++ b/internal/client/input/b2.go
@@ -16,6 +16,7 @@ var _ InputClient = (*B2InputClient)(nil)
 type B2InputClient struct {
 	prefix          string
 	bucket          *b2.Bucket
+	bucketName      string
 	b2cl            *b2.Client
 	knownExtensions []string
 }
@@ -36,7 +37,7 @@ func NewB2InputClient(cfg *config.InputConfig) (InputClient, error) {
 		return nil, err
 	}
 
-	return &B2InputClient{b2cl: b2cl, bucket: bucket, prefix: b2cfg.Prefix, knownExtensions: cfg.KnownExtensions}, nil
+	return &B2InputClient{b2cl: b2cl, bucket: bucket, bucketName: b2cfg.BucketName, prefix: b2cfg.Prefix, knownExtensions: cfg.KnownExtensions}, nil
 }
 
 func (c *B2InputClient) Scan() ([]string, error) {
@@ -120,6 +121,10 @@ func (c *B2InputClient) ReadMetadata(path string) (*MetadataStruct, error) {
 	return &metadata, nil
 }
 
+func (c *B2InputClient) ID(path string) string {
+	return fmt.Sprintf("b2://%s/%s%s", c.bucketName, c.prefix, path)
+}
+
 func (c *B2InputClient) GetReader(path string) (io.ReadCloser, error) {
 	obj := c.bucket.Object(c.prefix + path)
 	if obj == nil {
diff --git a/internal/client/input/input_client_interface.go b/internal/client/input/input_client_interface.go
index 4930aa0..a2cf84b 100644
--- a/internal/client/input/input_client_interface.go
+++ b/internal/client/input/input_client_interface.go
@@ -11,6 +11,7 @@ type InputClient interface {
 	Scan() ([]string, error)
 	ReadMetadata(string) (*MetadataStruct, error)
 	GetReader(string) (io.ReadCloser, error)
+	ID(path string) string
 }
 
 type MetadataStruct struct {
diff --git a/internal/client/input/local_unix.go b/internal/client/input/local_unix.go
index e509c8b..21b08a9 100644
--- a/internal/client/input/local_unix.go
+++ b/internal/client/input/local_unix.go
@@ -99,6 +99,10 @@ func (c *LocalUnixInputClient) ReadMetadata(path string) (*MetadataStruct, error
 	}, nil
 }
 
+func (c *LocalUnixInputClient) ID(path string) string {
+	return fmt.Sprintf("local-unix://%s%s", c.path, path)
+}
+
 func (c *LocalUnixInputClient) GetReader(path string) (io.ReadCloser, error) {
 	return os.Open(c.path + path)
 }
diff --git a/main.go b/main.go
index 4994116..c3f03e7 100644
--- a/main.go
+++ b/main.go
@@ -2,10 +2,15 @@ package main
 
 import (
 	"bytes"
+	"encoding/csv"
+	"encoding/json"
 	"flag"
+	"hash/crc32"
+	"io"
 	"log/slog"
 	"os"
 	"os/signal"
+	"strconv"
 	"strings"
 	"sync"
 	"syscall"
@@ -18,6 +23,7 @@ import (
 var (
 	configPath  = flag.String("c", "config.json", "Path to the configuration file")
 	sigTermChan = make(chan os.Signal, 1)
+	cacheMap    = make(map[string]map[uint32]struct{})
 )
 
 func main() {
@@ -48,8 +54,10 @@ func main() {
 	}
 
 	converters := make([]converter.Converter, 0, len(cfg.Converters))
-	var converterTypes []string
+	converterTypes := make([]string, 0, len(cfg.Converters))
+	converterHashes := make([]uint32, 0, len(cfg.Converters))
 	for _, converterCfg := range cfg.Converters {
+		converterBytes, _ := json.Marshal(converterCfg)
 		conv, err := converter.NewConverterMap[converterCfg.Type](&converterCfg)
 		if err != nil {
 			slog.Error("fail to initialize converter", slog.String("error", err.Error()))
@@ -57,6 +65,7 @@ func main() {
 		}
 		converters = append(converters, conv)
 		converterTypes = append(converterTypes, converterCfg.Type)
+		converterHashes = append(converterHashes, crc32.ChecksumIEEE(converterBytes))
 	}
 
 	generalLogger := slog.With(slog.String("input_storage", cfg.Input.Storage.Type))
@@ -84,6 +93,40 @@ func main() {
 	default:
 	}
 
+	if cfg.Input.CacheProcessed {
+		cacheFile, err := os.OpenFile(cfg.Input.CacheProcessedCsvPath, os.O_CREATE|os.O_RDONLY, 0644)
+		if err != nil {
+			generalLogger.Error("fail to initialize cache file", slog.String("cache_path", cfg.Input.CacheProcessedCsvPath), slog.String("error", err.Error()))
+			os.Exit(1)
+		}
+		defer cacheFile.Close()
+
+		csvReader := csv.NewReader(cacheFile)
+		for {
+			rec, err := csvReader.Read()
+			if err == io.EOF {
+				break
+			}
+			if err != nil {
+				generalLogger.Error("fail to initialize cache while reading file", slog.String("cache_path", cfg.Input.CacheProcessedCsvPath), slog.String("error", err.Error()))
+				os.Exit(1)
+			}
+			if len(rec) != 2 {
+				generalLogger.Error("fail to initialize cache while reading file", slog.Int("record_length", len(rec)), slog.String("error", "incorrect format: expected '{image-name},{semicolon-separated-processor-hashes}'"))
+				os.Exit(1)
+			}
+			hashes := strings.Split(rec[1], ";")
+			cacheMap[rec[0]] = make(map[uint32]struct{})
+			for _, hash := range hashes {
+				hashUint, err := strconv.ParseUint(hash, 10, 32)
+				if err != nil {
+					generalLogger.Error("fail to initialize cache while reading file", slog.Int("record_length", len(rec)), slog.String("error", err.Error()))
+				}
+				cacheMap[rec[0]][uint32(hashUint)] = struct{}{}
+			}
+		}
+	}
+
 	processSemaphore := make(chan struct{}, cfg.MaxProcessThreads)
 	queueSemaphore := make(chan struct{}, cfg.MaxPreProcessThreads)
 	var wg sync.WaitGroup
@@ -106,18 +149,36 @@ func main() {
 			default:
 			}
 
-			inputMetadata, err := inputClient.ReadMetadata(inputName)
-			if err != nil {
-				fileLogger.Warn("fail to read metadata of (supposedly existing) input file", slog.String("error", err.Error()))
-				return
+			var inputMetadata *input.MetadataStruct
+
+			id := inputClient.ID(file)
+			if _, ok := cacheMap[id]; !ok {
+				cacheMap[id] = make(map[uint32]struct{})
 			}
 
 			convertersToLaunch := []int{}
 			for j, conv := range converters {
+				if cfg.Input.CacheProcessed {
+					if _, ok := cacheMap[id][converterHashes[j]]; ok {
+						fileLogger.Info("skip already processed file (based on cache file containing it and processor)",
+							slog.String("file_id", id),
+							slog.Uint64("conv_hash", uint64(converterHashes[j])),
+							slog.Int("conv_index", j))
+						continue
+					}
+				}
 				outputName := conv.DeductOutputPath(inputName)
 				originalInputHash := ""
 				convLogger := fileLogger.With(slog.String("output_path", outputName), slog.Int("conv_index", j))
 
+				if inputMetadata == nil {
+					inputMetadata, err = inputClient.ReadMetadata(inputName)
+					if err != nil {
+						fileLogger.Warn("fail to read metadata of (supposedly existing) input file", slog.String("error", err.Error()))
+						return
+					}
+				}
+
 				if !cfg.ForceRewrite && !conv.IsMissing(outputName) {
 					outputMetadata, err := conv.ReadMetadata(outputName)
 					if err != nil {
@@ -127,6 +188,7 @@ func main() {
 					originalInputHash = outputMetadata.HashOriginal
 					if inputMetadata.Hash == originalInputHash {
 						convLogger.Info("skip already processed file (based on equal hash)", slog.String("input_hash", inputMetadata.Hash))
+						cacheMap[id][converterHashes[j]] = struct{}{}
 						continue
 					}
 				}
@@ -165,6 +227,7 @@ func main() {
 				}
 
 				convLogger.Info("successfully processed file", slog.String("input_hash", inputMetadata.Hash))
+				cacheMap[id][converterHashes[convIndex]] = struct{}{}
 			}
 		}(i, file)
 	}
@@ -176,7 +239,28 @@ func main() {
 		generalLogger.Info("exiting due to termination signal")
 		os.Exit(130)
 	default:
-		generalLogger.Info("all files processed successfully, exiting")
-		os.Exit(0)
 	}
+
+	generalLogger.Info("all files processed successfully")
+
+	if cfg.Input.CacheProcessed {
+		generalLogger.Info("writing cache file")
+		cacheFile, err := os.OpenFile(cfg.Input.CacheProcessedCsvPath, os.O_TRUNC|os.O_CREATE|os.O_WRONLY, 0644)
+		if err != nil {
+			generalLogger.Error("error writing cache into file")
+		} else {
+			for id, hashes := range cacheMap {
+				var hashStrings = make([]string, 0, len(hashes))
+				for hash := range hashes {
+					hashStrings = append(hashStrings, strconv.FormatUint(uint64(hash), 10))
+				}
+				cacheFile.Write([]byte(id))
+				cacheFile.Write([]byte{','})
+				cacheFile.WriteString(strings.Join(hashStrings, ";"))
+				cacheFile.Write([]byte{'\n'})
+			}
+		}
+	}
+
+	os.Exit(0)
 }