From 158506595f697907d98f4a0036fe4e0fe1d46dc9 Mon Sep 17 00:00:00 2001 From: Alec Thomas Date: Mon, 5 Jan 2026 12:37:33 +1100 Subject: [PATCH] feat: start of server implementation Initially this just includes an API for creating, reading, and deleting cache objects via a HTTP API. Also added in-memory cache implementation, primarily for testing. --- .golangci.yml | 4 + Justfile | 7 +- bin/.gotestsum-1.13.0.pkg | 1 + bin/gotestsum | 1 + internal/cache/api.go | 26 +++- internal/cache/cachetest/suite.go | 194 +++++++++++++++++++++++++++ internal/cache/disk.go | 2 +- internal/cache/disk_test.go | 126 ++--------------- internal/cache/memory.go | 179 ++++++++++++++++++++++++ internal/cache/memory_test.go | 20 +++ internal/cache/remote/client.go | 145 ++++++++++++++++++++ internal/cache/remote/client_test.go | 34 +++++ internal/cache/remote/server.go | 124 +++++++++++++++++ internal/server/server.go | 40 ++++++ 14 files changed, 784 insertions(+), 119 deletions(-) create mode 120000 bin/.gotestsum-1.13.0.pkg create mode 120000 bin/gotestsum create mode 100644 internal/cache/cachetest/suite.go create mode 100644 internal/cache/memory.go create mode 100644 internal/cache/memory_test.go create mode 100644 internal/cache/remote/client.go create mode 100644 internal/cache/remote/client_test.go create mode 100644 internal/cache/remote/server.go create mode 100644 internal/server/server.go diff --git a/.golangci.yml b/.golangci.yml index aff82713..d6392d5c 100644 --- a/.golangci.yml +++ b/.golangci.yml @@ -432,6 +432,10 @@ linters: # Default: false os-temp-dir: true + wrapcheck: + ignore-package-globs: + - "github.com/alecthomas/errors" + exclusions: # Predefined exclusion rules. # Default: [] diff --git a/Justfile b/Justfile index c747eea4..39111495 100644 --- a/Justfile +++ b/Justfile @@ -1,9 +1,12 @@ +set positional-arguments := true +set shell := ["bash", "-c"] + _help: @just -l # Run tests test: - go test ./... + @gotestsum --hide-summary output,skipped --format-hide-empty-pkg ${CI:+--format github-actions} # Lint code lint: @@ -13,5 +16,5 @@ lint: # Format code fmt: just --unstable --fmt - git ls-files | grep '\.go$' | xargs gosimports -local github.com/block -w + golangci-lint fmt go mod tidy diff --git a/bin/.gotestsum-1.13.0.pkg b/bin/.gotestsum-1.13.0.pkg new file mode 120000 index 00000000..383f4511 --- /dev/null +++ b/bin/.gotestsum-1.13.0.pkg @@ -0,0 +1 @@ +hermit \ No newline at end of file diff --git a/bin/gotestsum b/bin/gotestsum new file mode 120000 index 00000000..dbb2c93e --- /dev/null +++ b/bin/gotestsum @@ -0,0 +1 @@ +.gotestsum-1.13.0.pkg \ No newline at end of file diff --git a/internal/cache/api.go b/internal/cache/api.go index 7a206eef..170e0165 100644 --- a/internal/cache/api.go +++ b/internal/cache/api.go @@ -31,9 +31,31 @@ func Register[Config any, C Cache](id string, factory Factory[Config, C]) { // Key represents a unique identifier for a cached object. type Key [32]byte +// ParseKey from its hex-encoded string form. +func ParseKey(key string) (Key, error) { + var k Key + return k, k.UnmarshalText([]byte(key)) +} + func NewKey(url string) Key { return Key(sha256.Sum256([]byte(url))) } -func (k Key) String() string { return hex.EncodeToString(k[:]) } +func (k *Key) String() string { return hex.EncodeToString(k[:]) } + +func (k *Key) UnmarshalText(text []byte) error { + bytes, err := hex.DecodeString(string(text)) + if err != nil { + return errors.WithStack(err) + } + if len(bytes) != len(*k) { + return errors.New("invalid key length") + } + copy(k[:], bytes) + return nil +} + +func (k *Key) MarshalText() ([]byte, error) { + return []byte(k.String()), nil +} // A Cache knows how to retrieve, create and delete objects from a cache. type Cache interface { @@ -44,6 +66,8 @@ type Cache interface { Open(ctx context.Context, key Key) (io.ReadCloser, error) // Create a new file in the cache. // + // If "ttl" is zero, a maximum TTL MUST be used by the implementation. + // // The file MUST not be available for read until completely written and closed. Create(ctx context.Context, key Key, ttl time.Duration) (io.WriteCloser, error) // Delete a file from the cache. diff --git a/internal/cache/cachetest/suite.go b/internal/cache/cachetest/suite.go new file mode 100644 index 00000000..5ec72016 --- /dev/null +++ b/internal/cache/cachetest/suite.go @@ -0,0 +1,194 @@ +package cachetest + +import ( + "io" + "os" + "testing" + "time" + + "github.com/alecthomas/assert/v2" + + "github.com/block/sfptc/internal/cache" +) + +// Suite runs a comprehensive test suite against a cache.Cache implementation. +// All cache implementations should pass this test suite to ensure consistent semantics. +func Suite(t *testing.T, newCache func(t *testing.T) cache.Cache) { + t.Run("CreateAndOpen", func(t *testing.T) { + testCreateAndOpen(t, newCache(t)) + }) + + t.Run("NotFound", func(t *testing.T) { + testNotFound(t, newCache(t)) + }) + + t.Run("Expiration", func(t *testing.T) { + testExpiration(t, newCache(t)) + }) + + t.Run("DefaultTTL", func(t *testing.T) { + testDefaultTTL(t, newCache(t)) + }) + + t.Run("Delete", func(t *testing.T) { + testDelete(t, newCache(t)) + }) + + t.Run("MultipleWrites", func(t *testing.T) { + testMultipleWrites(t, newCache(t)) + }) + + t.Run("NotAvailableUntilClosed", func(t *testing.T) { + testNotAvailableUntilClosed(t, newCache(t)) + }) +} + +func testCreateAndOpen(t *testing.T, c cache.Cache) { + defer c.Close() + ctx := t.Context() + + key := cache.NewKey("test-key") + + writer, err := c.Create(ctx, key, time.Hour) + assert.NoError(t, err) + + _, err = writer.Write([]byte("hello world")) + assert.NoError(t, err) + + err = writer.Close() + assert.NoError(t, err) + + reader, err := c.Open(ctx, key) + assert.NoError(t, err) + defer reader.Close() + + data, err := io.ReadAll(reader) + assert.NoError(t, err) + assert.Equal(t, "hello world", string(data)) +} + +func testNotFound(t *testing.T, c cache.Cache) { + defer c.Close() + ctx := t.Context() + + key := cache.NewKey("nonexistent") + + _, err := c.Open(ctx, key) + assert.IsError(t, err, os.ErrNotExist) +} + +func testExpiration(t *testing.T, c cache.Cache) { + defer c.Close() + ctx := t.Context() + + key := cache.NewKey("test-key") + + writer, err := c.Create(ctx, key, 10*time.Millisecond) + assert.NoError(t, err) + + _, err = writer.Write([]byte("test data")) + assert.NoError(t, err) + + err = writer.Close() + assert.NoError(t, err) + + reader, err := c.Open(ctx, key) + assert.NoError(t, err) + assert.NoError(t, reader.Close()) + + time.Sleep(20 * time.Millisecond) + + _, err = c.Open(ctx, key) + assert.IsError(t, err, os.ErrNotExist) +} + +func testDefaultTTL(t *testing.T, c cache.Cache) { + defer c.Close() + ctx := t.Context() + + key := cache.NewKey("test-key") + + writer, err := c.Create(ctx, key, 0) + assert.NoError(t, err) + + _, err = writer.Write([]byte("test data")) + assert.NoError(t, err) + + err = writer.Close() + assert.NoError(t, err) + + reader, err := c.Open(ctx, key) + assert.NoError(t, err) + assert.NoError(t, reader.Close()) +} + +func testDelete(t *testing.T, c cache.Cache) { + defer c.Close() + ctx := t.Context() + + key := cache.NewKey("test-key") + + writer, err := c.Create(ctx, key, time.Hour) + assert.NoError(t, err) + + _, err = writer.Write([]byte("test data")) + assert.NoError(t, err) + + err = writer.Close() + assert.NoError(t, err) + + err = c.Delete(ctx, key) + assert.NoError(t, err) + + _, err = c.Open(ctx, key) + assert.IsError(t, err, os.ErrNotExist) +} + +func testMultipleWrites(t *testing.T, c cache.Cache) { + defer c.Close() + ctx := t.Context() + + key := cache.NewKey("test-key") + + writer, err := c.Create(ctx, key, time.Hour) + assert.NoError(t, err) + + _, err = writer.Write([]byte("hello ")) + assert.NoError(t, err) + + _, err = writer.Write([]byte("world")) + assert.NoError(t, err) + + err = writer.Close() + assert.NoError(t, err) + + reader, err := c.Open(ctx, key) + assert.NoError(t, err) + defer reader.Close() + + data, err := io.ReadAll(reader) + assert.NoError(t, err) + assert.Equal(t, "hello world", string(data)) +} + +func testNotAvailableUntilClosed(t *testing.T, c cache.Cache) { + defer c.Close() + ctx := t.Context() + + key := cache.NewKey("test-key") + + writer, err := c.Create(ctx, key, time.Hour) + assert.NoError(t, err) + + _, err = writer.Write([]byte("test data")) + assert.NoError(t, err) + + _, err = c.Open(ctx, key) + assert.IsError(t, err, os.ErrNotExist) + + err = writer.Close() + assert.NoError(t, err) + + _, err = c.Open(ctx, key) + assert.NoError(t, err) +} diff --git a/internal/cache/disk.go b/internal/cache/disk.go index a16d716c..21df964b 100644 --- a/internal/cache/disk.go +++ b/internal/cache/disk.go @@ -27,7 +27,7 @@ func init() { type DiskConfig struct { Root string `hcl:"root" help:"Root directory for the disk storage."` - LimitMB int `hcl:"limit-mb,optional" help:"Maximum size of the disk cache in megabytes (defaults to 1GB)." default:"1024"` + LimitMB int `hcl:"limit-mb,optional" help:"Maximum size of the disk cache in megabytes (defaults to 10GB)." default:"10240"` MaxTTL time.Duration `hcl:"max-ttl,optional" help:"Maximum time-to-live for entries in the disk cache (defaults to 1 hour)." default:"1h"` EvictInterval time.Duration `hcl:"evict-interval,optional" help:"Interval at which to check files for eviction (defaults to 1 minute)." default:"1m"` } diff --git a/internal/cache/disk_test.go b/internal/cache/disk_test.go index 758ebcf5..fbbce87f 100644 --- a/internal/cache/disk_test.go +++ b/internal/cache/disk_test.go @@ -2,129 +2,25 @@ package cache_test import ( "log/slog" - "os" - "path/filepath" "testing" "time" "github.com/alecthomas/assert/v2" "github.com/block/sfptc/internal/cache" + "github.com/block/sfptc/internal/cache/cachetest" "github.com/block/sfptc/internal/logging" ) -func testKey(s string) cache.Key { - return cache.NewKey(s) -} - -func TestDiskStorageExpiry(t *testing.T) { - dir := t.TempDir() - _, ctx := logging.Configure(t.Context(), logging.Config{Level: slog.LevelDebug}) - - disk, err := cache.NewDisk(ctx, cache.DiskConfig{Root: dir}) - assert.NoError(t, err) - - assert.Equal(t, int64(0), disk.Size()) - - key1 := testKey("first") - key2 := testKey("second") - - w1, err := disk.Create(ctx, key1, time.Second*3) - assert.NoError(t, err) - _, err = w1.Write([]byte("first file data")) - assert.NoError(t, err) - assert.NoError(t, w1.Close()) - - w2, err := disk.Create(ctx, key2, time.Second) - assert.NoError(t, err) - _, err = w2.Write([]byte("second file data")) - assert.NoError(t, err) - assert.NoError(t, w2.Close()) - - expectedSize := int64(len("first file data") + len("second file data")) - assert.Equal(t, expectedSize, disk.Size()) - - time.Sleep(time.Second * 2) - - r, err := disk.Open(ctx, key1) - assert.NoError(t, err) - assert.NoError(t, r.Close()) - - _, err = disk.Open(ctx, key2) - assert.IsError(t, err, os.ErrNotExist) - - assert.Equal(t, int64(len("first file data")), disk.Size()) -} - -func TestDiskAsyncEviction(t *testing.T) { - dir := t.TempDir() - _, ctx := logging.Configure(t.Context(), logging.Config{Level: slog.LevelDebug}) - - disk, err := cache.NewDisk(ctx, cache.DiskConfig{ - Root: dir, - EvictInterval: 500 * time.Millisecond, +func TestDiskCache(t *testing.T) { + cachetest.Suite(t, func(t *testing.T) cache.Cache { + dir := t.TempDir() + _, ctx := logging.Configure(t.Context(), logging.Config{Level: slog.LevelDebug}) + c, err := cache.NewDisk(ctx, cache.DiskConfig{ + Root: dir, + MaxTTL: 100 * time.Millisecond, + }) + assert.NoError(t, err) + return c }) - assert.NoError(t, err) - t.Cleanup(func() { assert.NoError(t, disk.Close()) }) - - assert.Equal(t, int64(0), disk.Size(), "initial size should be 0") - - keyExpire1 := testKey("expire1") - keyExpire2 := testKey("expire2") - keyKeep1 := testKey("keep1") - keyKeep2 := testKey("keep2") - - // Create files that will expire - w1, err := disk.Create(ctx, keyExpire1, 800*time.Millisecond) - assert.NoError(t, err) - _, err = w1.Write([]byte("expires soon")) - assert.NoError(t, err) - assert.NoError(t, w1.Close()) - - w2, err := disk.Create(ctx, keyExpire2, 800*time.Millisecond) - assert.NoError(t, err) - _, err = w2.Write([]byte("expires soon")) - assert.NoError(t, err) - assert.NoError(t, w2.Close()) - - // Create files that won't expire - w3, err := disk.Create(ctx, keyKeep1, 10*time.Second) - assert.NoError(t, err) - _, err = w3.Write([]byte("keep this")) - assert.NoError(t, err) - assert.NoError(t, w3.Close()) - - w4, err := disk.Create(ctx, keyKeep2, 10*time.Second) - assert.NoError(t, err) - _, err = w4.Write([]byte("keep this too")) - assert.NoError(t, err) - assert.NoError(t, w4.Close()) - - expectedSize := int64(len("expires soon")*2 + len("keep this") + len("keep this too")) - assert.Equal(t, expectedSize, disk.Size(), "size should match total written data") - - // Wait for expired files to be evicted - time.Sleep(1500 * time.Millisecond) - - // Check filesystem directly to verify async eviction actually deleted files - hexExpire1 := keyExpire1.String() - _, err = os.Stat(filepath.Join(dir, hexExpire1[:2], hexExpire1)) - assert.IsError(t, err, os.ErrNotExist, "expire1 should be deleted from disk") - - hexExpire2 := keyExpire2.String() - _, err = os.Stat(filepath.Join(dir, hexExpire2[:2], hexExpire2)) - assert.IsError(t, err, os.ErrNotExist, "expire2 should be deleted from disk") - - // Non-expired files should still exist on disk - hexKeep1 := keyKeep1.String() - _, err = os.Stat(filepath.Join(dir, hexKeep1[:2], hexKeep1)) - assert.NoError(t, err, "keep1 should still exist on disk") - - hexKeep2 := keyKeep2.String() - _, err = os.Stat(filepath.Join(dir, hexKeep2[:2], hexKeep2)) - assert.NoError(t, err, "keep2 should still exist on disk") - - // Verify size only includes non-expired files - expectedSizeAfterEviction := int64(len("keep this") + len("keep this too")) - assert.Equal(t, expectedSizeAfterEviction, disk.Size(), "size should only include non-expired files") } diff --git a/internal/cache/memory.go b/internal/cache/memory.go new file mode 100644 index 00000000..3ed98447 --- /dev/null +++ b/internal/cache/memory.go @@ -0,0 +1,179 @@ +package cache + +import ( + "bytes" + "context" + "io" + "os" + "sync" + "time" + + "github.com/alecthomas/errors" +) + +func init() { + Register("memory", NewMemoryCache) +} + +type MemoryCacheConfig struct { + LimitMB int `hcl:"limit-mb,optional" help:"Maximum size of the disk cache in megabytes (defaults to 1GB)." default:"1024"` + MaxTTL time.Duration `hcl:"max-ttl,optional" help:"Maximum time-to-live for entries in the disk cache (defaults to 1 hour)." default:"1h"` +} + +type memoryEntry struct { + data []byte + expiresAt time.Time +} + +type memoryCache struct { + config MemoryCacheConfig + mu sync.RWMutex + entries map[Key]*memoryEntry + currentSize int64 +} + +func NewMemoryCache(_ context.Context, config MemoryCacheConfig) (Cache, error) { + return &memoryCache{ + config: config, + entries: make(map[Key]*memoryEntry), + }, nil +} + +func (m *memoryCache) Open(_ context.Context, key Key) (io.ReadCloser, error) { + m.mu.RLock() + defer m.mu.RUnlock() + + entry, exists := m.entries[key] + if !exists { + return nil, os.ErrNotExist + } + + if time.Now().After(entry.expiresAt) { + return nil, os.ErrNotExist + } + + return io.NopCloser(bytes.NewReader(entry.data)), nil +} + +func (m *memoryCache) Create(_ context.Context, key Key, ttl time.Duration) (io.WriteCloser, error) { + if ttl == 0 { + ttl = m.config.MaxTTL + } + + writer := &memoryWriter{ + cache: m, + key: key, + buf: &bytes.Buffer{}, + expiresAt: time.Now().Add(ttl), + } + + return writer, nil +} + +func (m *memoryCache) Delete(_ context.Context, key Key) error { + m.mu.Lock() + defer m.mu.Unlock() + + entry, exists := m.entries[key] + if !exists { + return os.ErrNotExist + } + m.currentSize -= int64(len(entry.data)) + delete(m.entries, key) + return nil +} + +func (m *memoryCache) Close() error { + m.mu.Lock() + defer m.mu.Unlock() + + m.entries = nil + return nil +} + +type memoryWriter struct { + cache *memoryCache + key Key + buf *bytes.Buffer + expiresAt time.Time + closed bool +} + +func (w *memoryWriter) Write(p []byte) (int, error) { + if w.closed { + return 0, errors.New("writer closed") + } + return errors.WithStack2(w.buf.Write(p)) +} + +func (w *memoryWriter) Close() error { + if w.closed { + return nil + } + w.closed = true + + w.cache.mu.Lock() + defer w.cache.mu.Unlock() + + newSize := int64(w.buf.Len()) + limitBytes := int64(w.cache.config.LimitMB) * 1024 * 1024 + + // Remove old entry size if it exists + oldSize := int64(0) + if oldEntry, exists := w.cache.entries[w.key]; exists { + oldSize = int64(len(oldEntry.data)) + } + + // Evict entries if needed to make room + if limitBytes > 0 { + neededSpace := w.cache.currentSize - oldSize + newSize - limitBytes + if neededSpace > 0 { + w.cache.evictOldest(neededSpace) + } + } + + w.cache.currentSize -= oldSize + w.cache.entries[w.key] = &memoryEntry{ + data: w.buf.Bytes(), + expiresAt: w.expiresAt, + } + w.cache.currentSize += newSize + + return nil +} + +func (m *memoryCache) evictOldest(neededSpace int64) { + type entryInfo struct { + key Key + size int64 + expiresAt time.Time + } + + var entries []entryInfo + for k, e := range m.entries { + entries = append(entries, entryInfo{ + key: k, + size: int64(len(e.data)), + expiresAt: e.expiresAt, + }) + } + + // Sort by expiry time (earliest first) + for i := 0; i < len(entries); i++ { + for j := i + 1; j < len(entries); j++ { + if entries[i].expiresAt.After(entries[j].expiresAt) { + entries[i], entries[j] = entries[j], entries[i] + } + } + } + + freedSpace := int64(0) + for _, e := range entries { + if freedSpace >= neededSpace { + break + } + m.currentSize -= e.size + delete(m.entries, e.key) + freedSpace += e.size + } +} diff --git a/internal/cache/memory_test.go b/internal/cache/memory_test.go new file mode 100644 index 00000000..730466a0 --- /dev/null +++ b/internal/cache/memory_test.go @@ -0,0 +1,20 @@ +package cache_test + +import ( + "testing" + "time" + + "github.com/alecthomas/assert/v2" + + "github.com/block/sfptc/internal/cache" + "github.com/block/sfptc/internal/cache/cachetest" +) + +func TestMemoryCache(t *testing.T) { + cachetest.Suite(t, func(t *testing.T) cache.Cache { + ctx := t.Context() + c, err := cache.NewMemoryCache(ctx, cache.MemoryCacheConfig{MaxTTL: 100 * time.Millisecond}) + assert.NoError(t, err) + return c + }) +} diff --git a/internal/cache/remote/client.go b/internal/cache/remote/client.go new file mode 100644 index 00000000..139cd8e0 --- /dev/null +++ b/internal/cache/remote/client.go @@ -0,0 +1,145 @@ +package remote + +import ( + "context" + "fmt" + "io" + "net/http" + "os" + "time" + + "github.com/alecthomas/errors" + + "github.com/block/sfptc/internal/cache" +) + +// Client implements cache.Cache as a client for the remote cache server. +type Client struct { + baseURL string + client *http.Client +} + +var _ cache.Cache = (*Client)(nil) + +// NewClient creates a new remote cache client. +func NewClient(baseURL string) *Client { + return &Client{ + baseURL: baseURL, + client: &http.Client{}, + } +} + +// Open retrieves an object from the remote cache. +func (c *Client) Open(ctx context.Context, key cache.Key) (io.ReadCloser, error) { + url := fmt.Sprintf("%s/%s", c.baseURL, key.String()) + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, errors.Wrap(err, "failed to create request") + } + + resp, err := c.client.Do(req) + if err != nil { + return nil, errors.Wrap(err, "failed to execute request") + } + + if resp.StatusCode == http.StatusNotFound { + return nil, errors.Join(os.ErrNotExist, resp.Body.Close()) + } + + if resp.StatusCode != http.StatusOK { + return nil, errors.Join(errors.Errorf("unexpected status code: %d", resp.StatusCode), resp.Body.Close()) + } + + return resp.Body, nil +} + +// Create stores a new object in the remote cache. +func (c *Client) Create(ctx context.Context, key cache.Key, ttl time.Duration) (io.WriteCloser, error) { + pr, pw := io.Pipe() + + url := fmt.Sprintf("%s/%s", c.baseURL, key.String()) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, url, pr) + if err != nil { + return nil, errors.Join(errors.Wrap(err, "failed to create request"), pr.Close(), pw.Close()) + } + + if ttl > 0 { + req.Header.Set("Time-To-Live", ttl.String()) + } + + wc := &writeCloser{ + pw: pw, + done: make(chan error, 1), + } + + go func() { + resp, err := c.client.Do(req) + if err != nil { + wc.done <- errors.Wrap(err, "failed to execute request") + return + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + wc.done <- errors.Errorf("unexpected status code: %d", resp.StatusCode) + return + } + + wc.done <- nil + }() + + return wc, nil +} + +// Delete removes an object from the remote cache. +func (c *Client) Delete(ctx context.Context, key cache.Key) error { + url := fmt.Sprintf("%s/%s", c.baseURL, key.String()) + req, err := http.NewRequestWithContext(ctx, http.MethodDelete, url, nil) + if err != nil { + return errors.Wrap(err, "failed to create request") + } + + resp, err := c.client.Do(req) + if err != nil { + return errors.Wrap(err, "failed to execute request") + } + defer resp.Body.Close() + + if resp.StatusCode == http.StatusNotFound { + return os.ErrNotExist + } + + if resp.StatusCode != http.StatusOK { + return errors.Errorf("unexpected status code: %d", resp.StatusCode) + } + + return nil +} + +// Close closes the client and releases resources. +func (c *Client) Close() error { + c.client.CloseIdleConnections() + return nil +} + +// writeCloser wraps a pipe writer and waits for the HTTP request to complete. +type writeCloser struct { + pw *io.PipeWriter + done chan error +} + +func (wc *writeCloser) Write(p []byte) (int, error) { + n, err := wc.pw.Write(p) + return n, errors.WithStack(err) +} + +func (wc *writeCloser) Close() error { + if err := wc.pw.Close(); err != nil { + return errors.Wrap(err, "failed to close pipe writer") + } + err := <-wc.done + if err != nil { + return errors.Wrap(err, "request failed") + } + return nil +} diff --git a/internal/cache/remote/client_test.go b/internal/cache/remote/client_test.go new file mode 100644 index 00000000..8a059246 --- /dev/null +++ b/internal/cache/remote/client_test.go @@ -0,0 +1,34 @@ +package remote_test + +import ( + "log/slog" + "net/http/httptest" + "testing" + "time" + + "github.com/alecthomas/assert/v2" + + "github.com/block/sfptc/internal/cache" + "github.com/block/sfptc/internal/cache/cachetest" + "github.com/block/sfptc/internal/cache/remote" + "github.com/block/sfptc/internal/logging" +) + +func TestRemoteClient(t *testing.T) { + cachetest.Suite(t, func(t *testing.T) cache.Cache { + ctx := t.Context() + _, ctx = logging.Configure(ctx, logging.Config{Level: slog.LevelError}) + memCache, err := cache.NewMemoryCache(ctx, cache.MemoryCacheConfig{ + MaxTTL: 100 * time.Millisecond, + }) + assert.NoError(t, err) + t.Cleanup(func() { memCache.Close() }) + + server := remote.NewServer(ctx, memCache) + ts := httptest.NewServer(server) + t.Cleanup(ts.Close) + + client := remote.NewClient(ts.URL) + return client + }) +} diff --git a/internal/cache/remote/server.go b/internal/cache/remote/server.go new file mode 100644 index 00000000..b02f9cec --- /dev/null +++ b/internal/cache/remote/server.go @@ -0,0 +1,124 @@ +// Package remote provides the server and client for a remote [cache.Cache] implementation. +package remote + +import ( + "context" + "errors" + "io" + "log/slog" + "net/http" + "os" + "time" + + "github.com/block/sfptc/internal/cache" + "github.com/block/sfptc/internal/logging" +) + +// Server side implementation of the cache protocol. +type Server struct { + cache cache.Cache + logger *slog.Logger + mux *http.ServeMux +} + +var _ http.Handler = (*Server)(nil) + +func NewServer(ctx context.Context, cache cache.Cache) *Server { + s := &Server{ + logger: logging.FromContext(ctx), + cache: cache, + mux: http.NewServeMux(), + } + s.mux.Handle("GET /{key}", http.HandlerFunc(s.getObject)) + s.mux.Handle("POST /{key}", http.HandlerFunc(s.putObject)) + s.mux.Handle("DELETE /{key}", http.HandlerFunc(s.deleteObject)) + return s +} + +func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { + s.mux.ServeHTTP(w, r) +} + +func (s *Server) getObject(w http.ResponseWriter, r *http.Request) { + key, err := cache.ParseKey(r.PathValue("key")) + if err != nil { + s.httpError(w, http.StatusBadRequest, err, "Invalid key") + return + } + + cr, err := s.cache.Open(r.Context(), key) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + s.httpError(w, http.StatusNotFound, err, "Cache object not found", slog.String("key", key.String())) + return + } + s.httpError(w, http.StatusInternalServerError, err, "Failed to open cache object", slog.String("key", key.String())) + return + } + + _, err = io.Copy(w, cr) + if err != nil { + s.logger.Error("Failed to copy cache object to response", slog.String("error", err.Error()), slog.String("key", key.String())) + } + if cerr := cr.Close(); cerr != nil { + s.logger.Error("Failed to close cache reader", slog.String("error", cerr.Error()), slog.String("key", key.String())) + } +} + +func (s *Server) putObject(w http.ResponseWriter, r *http.Request) { + key, err := cache.ParseKey(r.PathValue("key")) + if err != nil { + s.httpError(w, http.StatusBadRequest, err, "Invalid key") + return + } + + var ttl time.Duration + ttlh := r.Header.Get("Time-To-Live") + if ttlh != "" { + ttl, err = time.ParseDuration(ttlh) + if err != nil { + s.httpError(w, http.StatusBadRequest, err, "Invalid Time-To-Live header format, must be in Go duration format eg. 1h") + return + } + } + + cw, err := s.cache.Create(r.Context(), key, ttl) + if err != nil { + s.httpError(w, http.StatusInternalServerError, err, "Failed to create cache writer", slog.String("key", key.String())) + return + } + + if _, err := io.Copy(cw, r.Body); err != nil { + s.httpError(w, http.StatusInternalServerError, err, "Failed to copy request body to cache writer") + return + } + + if err := cw.Close(); err != nil { + s.httpError(w, http.StatusInternalServerError, err, "Failed to close cache writer") + return + } +} + +func (s *Server) deleteObject(w http.ResponseWriter, r *http.Request) { + key, err := cache.ParseKey(r.PathValue("key")) + if err != nil { + s.httpError(w, http.StatusBadRequest, err, "Invalid key") + return + } + + err = s.cache.Delete(r.Context(), key) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + s.httpError(w, http.StatusNotFound, err, "Cache object not found", slog.String("key", key.String())) + return + } + s.httpError(w, http.StatusInternalServerError, err, "Failed to delete cache object", slog.String("key", key.String())) + return + } +} + +func (s *Server) httpError(w http.ResponseWriter, code int, err error, message string, args ...any) { + args = append(args, slog.String("error", err.Error())) + s.logger.Error(message, args...) + http.Error(w, message, code) +} diff --git a/internal/server/server.go b/internal/server/server.go new file mode 100644 index 00000000..43da9f6a --- /dev/null +++ b/internal/server/server.go @@ -0,0 +1,40 @@ +// Package server implements the HTTP server for the caching proxy. +package server + +import ( + "context" + "log/slog" + "net/http" + + "github.com/block/sfptc/internal/cache" + "github.com/block/sfptc/internal/cache/remote" + "github.com/block/sfptc/internal/logging" +) + +type Option func(*Server) + +type Server struct { + logger *slog.Logger + cache cache.Cache + mux *http.ServeMux + server *remote.Server +} + +var _ http.Handler = (*Server)(nil) + +func New(ctx context.Context, cache cache.Cache, options ...Option) *Server { + s := &Server{ + logger: logging.FromContext(ctx), + cache: cache, + mux: http.NewServeMux(), + server: remote.NewServer(ctx, cache), + } + for _, option := range options { + option(s) + } + return s +} + +func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { + s.mux.ServeHTTP(w, r) +}