feat: rewrite S3 metadata backend as op journal

cmd/cachewd/main.go

@@ -30,6 +30,7 @@ import (
 	"github.com/block/cachew/internal/jobscheduler"
 	"github.com/block/cachew/internal/logging"
 	"github.com/block/cachew/internal/metadatadb"
+	metadatas3 "github.com/block/cachew/internal/metadatadb/s3"
 	"github.com/block/cachew/internal/metrics"
 	"github.com/block/cachew/internal/opa"
 	"github.com/block/cachew/internal/reaper"
@@ -260,7 +261,7 @@ func newRegistries(
 
 	mr := metadatadb.NewRegistry()
 	metadatadb.RegisterMemory(mr)
-	metadatadb.RegisterS3(mr, s3ClientProvider)
+	metadatas3.Register(mr, s3ClientProvider)
 
 	sr := strategy.NewRegistry()
 	strategy.RegisterAPIV1(sr)

internal/metadatadb/ops.go

@@ -7,10 +7,6 @@ import (
 	"github.com/alecthomas/errors"
 )
 
-// errInvalidToken is returned when an optimistic concurrency token does not
-// match the current version, indicating a concurrent write.
-var errInvalidToken = errors.New("invalid token")
-
 // Op is a write operation applied to the metadata store. Concrete types form a
 // closed set (sum type) — backends handle each variant via exhaustive type switch.
 //
@@ -238,6 +234,17 @@ type ListLen struct{ Key string }
 
 func (ListLen) readOp() {}
 
+// ApplyOp applies a single write Op to raw namespace state. It is exported
+// for Backend implementations that maintain state outside this package.
+func ApplyOp(state map[string]any, o Op) { applyOp(state, o) }
+
+// QueryStateInto executes a read query against raw namespace state and
+// unmarshals the result into target. It is exported for Backend
+// implementations that maintain state outside this package.
+func QueryStateInto(state map[string]any, q ReadOp, target any) error {
+	return errors.Wrap(jsonUnmarshalInto(queryState(state, q), target), "query state")
+}
+
 // applyOp applies a single write Op to the in-memory state via exhaustive type switch.
 func applyOp(state map[string]any, o Op) { //nolint:funlen
 	switch o := o.(type) {