diff --git a/auth/apikey.go b/auth/apikey.go new file mode 100644 index 0000000..21e2891 --- /dev/null +++ b/auth/apikey.go @@ -0,0 +1,166 @@ +package auth + +import ( + "crypto/rand" + "crypto/sha256" + "crypto/subtle" + "encoding/base64" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "net/http" + "os" + "path/filepath" + "sync" + "time" +) + +// APIKey is the stored record for a minted key. Only the SHA-256 of the +// full key string is kept; the key itself is shown exactly once at mint +// time. SHA-256 (not argon2/bcrypt) is sufficient because keys are 192-bit +// random secrets, not low-entropy passwords. +type APIKey struct { + ID string `json:"id"` + Name string `json:"name"` + Hash string `json:"hash"` // hex sha256 of the full key string + Scopes []string `json:"scopes,omitempty"` + CreatedAt time.Time `json:"createdAt"` +} + +// ErrKeyNotFound is returned by Revoke for an unknown key id. +var ErrKeyNotFound = errors.New("auth: api key not found") + +// KeyStore is a file-backed set of API keys ( is JSON, written +// atomically). Volume is tiny — a handful of integration keys — so a flat +// file beats dragging key records into the event-sourced graph store. +type KeyStore struct { + path string + mu sync.Mutex + keys []APIKey +} + +// OpenKeyStore loads (or initializes) the key store at path. +func OpenKeyStore(path string) (*KeyStore, error) { + s := &KeyStore{path: path} + raw, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return s, nil + } + if err != nil { + return nil, err + } + if err := json.Unmarshal(raw, &s.keys); err != nil { + return nil, fmt.Errorf("auth: parse %s: %w", path, err) + } + return s, nil +} + +// Mint creates a key with the given display name and scopes, persists its +// hash, and returns the full key string — the only time it is available. +func (s *KeyStore) Mint(name string, scopes []string) (key string, rec APIKey, err error) { + idBytes := make([]byte, 4) + secret := make([]byte, 24) + if _, err = rand.Read(idBytes); err != nil { + return "", APIKey{}, err + } + if _, err = rand.Read(secret); err != nil { + return "", APIKey{}, err + } + id := hex.EncodeToString(idBytes) + key = "gck_" + id + "_" + base64.RawURLEncoding.EncodeToString(secret) + rec = APIKey{ + ID: id, + Name: name, + Hash: hashKey(key), + Scopes: scopes, + CreatedAt: time.Now().UTC(), + } + s.mu.Lock() + defer s.mu.Unlock() + s.keys = append(s.keys, rec) + if err := s.saveLocked(); err != nil { + s.keys = s.keys[:len(s.keys)-1] + return "", APIKey{}, err + } + return key, rec, nil +} + +// Revoke deletes the key with the given id. +func (s *KeyStore) Revoke(id string) error { + s.mu.Lock() + defer s.mu.Unlock() + for i, k := range s.keys { + if k.ID == id { + s.keys = append(s.keys[:i], s.keys[i+1:]...) + return s.saveLocked() + } + } + return ErrKeyNotFound +} + +// List returns the stored key records. Callers serving them over HTTP +// should strip Hash themselves — it isn't secret-bearing, but is also not +// a client concern. +func (s *KeyStore) List() []APIKey { + s.mu.Lock() + defer s.mu.Unlock() + out := make([]APIKey, len(s.keys)) + copy(out, s.keys) + return out +} + +// lookup finds the record whose hash matches the presented key. +func (s *KeyStore) lookup(presented string) (APIKey, bool) { + h := []byte(hashKey(presented)) + s.mu.Lock() + defer s.mu.Unlock() + for _, k := range s.keys { + if subtle.ConstantTimeCompare(h, []byte(k.Hash)) == 1 { + return k, true + } + } + return APIKey{}, false +} + +// saveLocked writes the store atomically (tmp + rename). Callers hold s.mu. +func (s *KeyStore) saveLocked() error { + raw, err := json.MarshalIndent(s.keys, "", " ") + if err != nil { + return err + } + if err := os.MkdirAll(filepath.Dir(s.path), 0o755); err != nil { + return err + } + tmp := s.path + ".tmp" + if err := os.WriteFile(tmp, raw, 0o600); err != nil { + return err + } + return os.Rename(tmp, s.path) +} + +func hashKey(key string) string { + sum := sha256.Sum256([]byte(key)) + return hex.EncodeToString(sum[:]) +} + +// APIKeyAuth authenticates requests carrying `X-API-Key: ` against a +// KeyStore. The Principal's Subject is the key id and its Scopes are the +// scopes the key was minted with. +type APIKeyAuth struct { + Keys *KeyStore +} + +var _ Authenticator = APIKeyAuth{} + +func (a APIKeyAuth) Authenticate(r *http.Request) (*Principal, error) { + presented := r.Header.Get("X-API-Key") + if presented == "" { + return nil, ErrNoCredentials + } + rec, ok := a.Keys.lookup(presented) + if !ok { + return nil, errors.New("auth: invalid api key") + } + return &Principal{Subject: rec.ID, Scopes: rec.Scopes, Kind: KindAPIKey}, nil +} diff --git a/auth/auth.go b/auth/auth.go new file mode 100644 index 0000000..d8d47a4 --- /dev/null +++ b/auth/auth.go @@ -0,0 +1,87 @@ +// Package auth is the stable cross-service authentication seam for the +// platform. It is provider-agnostic: one Authenticator interface with +// interchangeable impls — OIDC (validates a bearer JWT against any OIDC +// issuer; Zitadel is the default reference), API key (validates X-API-Key +// against the KeyStore), and a Chain combinator that runs them in order. +// +// Packages under platform/ import only the standard library and other +// platform/ packages — never an application module. Application modules +// depend downward onto platform. This is what keeps the auth seam stable +// and reusable. +package auth + +import ( + "errors" + "net/http" + "slices" +) + +// Kind values for Principal.Kind. Comparing against these strings is the +// authorization layer's primitive: handlers ask "is this OIDC or an API +// key?" rather than re-implementing the chain logic. +const ( + KindOIDC = "oidc" + KindAPIKey = "apikey" + KindAnonymous = "anonymous" +) + +// ErrNoCredentials is returned when a request carries no credentials for the +// authenticator (no bearer token / no API key). Chain treats it like any +// other failure and falls through to the next authenticator. +var ErrNoCredentials = errors.New("auth: no credentials") + +// Principal is the authenticated caller. Authorization (scope checks) is a +// thin layer on top: handlers ask "does this principal have scope X", not +// "how did it authenticate". +type Principal struct { + Subject string `json:"subject"` // OIDC sub or API key id + Scopes []string `json:"scopes,omitempty"` // e.g. ["write","admin"] + Kind string `json:"kind"` // KindOIDC | KindAPIKey | KindAnonymous +} + +// HasScope reports whether the principal carries the given scope. +func (p *Principal) HasScope(scope string) bool { + return slices.Contains(p.Scopes, scope) +} + +// Authenticator inspects request credentials and returns the caller, or an +// error if authentication fails. +type Authenticator interface { + Authenticate(r *http.Request) (*Principal, error) +} + +// Chain tries each authenticator in order, returning the first successful +// Principal. It is how bearer-then-apikey composition is expressed. +type Chain []Authenticator + +var _ Authenticator = Chain(nil) + +// Authenticate returns the first non-error result. On failure it returns +// the first *real* validation error — an authenticator reporting +// ErrNoCredentials merely didn't apply (e.g. no X-API-Key header alongside +// a bad bearer token) and must not mask why the presented credential was +// rejected. +func (c Chain) Authenticate(r *http.Request) (*Principal, error) { + var firstErr error + for _, a := range c { + p, err := a.Authenticate(r) + if err == nil { + return p, nil + } + if firstErr == nil || (errors.Is(firstErr, ErrNoCredentials) && !errors.Is(err, ErrNoCredentials)) { + firstErr = err + } + } + return nil, firstErr +} + +// Anonymous authenticates every request as an unauthenticated principal +// with no scopes. It is the placeholder for tests and local-dev runs; it +// must never be wired in production. +type Anonymous struct{} + +var _ Authenticator = Anonymous{} + +func (Anonymous) Authenticate(*http.Request) (*Principal, error) { + return &Principal{Subject: "anonymous", Kind: KindAnonymous}, nil +} diff --git a/auth/auth_test.go b/auth/auth_test.go new file mode 100644 index 0000000..0f1d0be --- /dev/null +++ b/auth/auth_test.go @@ -0,0 +1,302 @@ +package auth + +import ( + "crypto" + "crypto/rand" + "crypto/rsa" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" + "time" +) + +// --- OIDC --- + +// fakeIssuer hosts a discovery document and a JWKS for a generated RSA key, +// and mints RS256 tokens signed by it. +type fakeIssuer struct { + srv *httptest.Server + key *rsa.PrivateKey +} + +func newFakeIssuer(t *testing.T) *fakeIssuer { + t.Helper() + key, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + t.Fatal(err) + } + f := &fakeIssuer{key: key} + mux := http.NewServeMux() + mux.HandleFunc("/.well-known/openid-configuration", func(w http.ResponseWriter, _ *http.Request) { + _ = json.NewEncoder(w).Encode(map[string]string{"jwks_uri": f.srv.URL + "/jwks"}) + }) + mux.HandleFunc("/jwks", func(w http.ResponseWriter, _ *http.Request) { + pub := &key.PublicKey + _ = json.NewEncoder(w).Encode(map[string]any{ + "keys": []map[string]string{{ + "kty": "RSA", + "kid": "test-key", + "n": base64.RawURLEncoding.EncodeToString(pub.N.Bytes()), + "e": base64.RawURLEncoding.EncodeToString(mustBigIntBytes(pub.E)), + }}, + }) + }) + f.srv = httptest.NewServer(mux) + t.Cleanup(f.srv.Close) + return f +} + +// mint signs an RS256 JWT with the given claims (kid "test-key"). +func (f *fakeIssuer) mint(t *testing.T, claims map[string]any) string { + t.Helper() + enc := func(v any) string { + b, err := json.Marshal(v) + if err != nil { + t.Fatal(err) + } + return base64.RawURLEncoding.EncodeToString(b) + } + signing := enc(map[string]string{"alg": "RS256", "kid": "test-key"}) + "." + enc(claims) + digest := sha256.Sum256([]byte(signing)) + sig, err := rsa.SignPKCS1v15(rand.Reader, f.key, crypto.SHA256, digest[:]) + if err != nil { + t.Fatal(err) + } + return signing + "." + base64.RawURLEncoding.EncodeToString(sig) +} + +// mustBigIntBytes turns an RSA exponent (typically 65537) into its +// big-endian byte encoding for the JWK "e" field. +func mustBigIntBytes(e int) []byte { + // smallest big-endian encoding of e. + n := e + if n == 0 { + return []byte{0} + } + var buf []byte + for n > 0 { + buf = append([]byte{byte(n & 0xff)}, buf...) + n >>= 8 + } + return buf +} + +func bearerReq(token string) *http.Request { + r := httptest.NewRequest("GET", "/api/nodes", nil) + r.Header.Set("Authorization", "Bearer "+token) + return r +} + +func TestOIDCAuthenticate(t *testing.T) { + iss := newFakeIssuer(t) + o := NewOIDC(iss.srv.URL, "my-client") + + valid := map[string]any{ + "iss": iss.srv.URL, + "sub": "user-1", + "email": "user-1@example.com", + "name": "User One", + "aud": "my-client", + "exp": time.Now().Add(time.Hour).Unix(), + "scope": "openid profile", + } + + p, err := o.Authenticate(bearerReq(iss.mint(t, valid))) + if err != nil { + t.Fatalf("valid token rejected: %v", err) + } + if p.Subject != "user-1" || p.Kind != KindOIDC || !p.HasScope("profile") { + t.Fatalf("unexpected principal: %+v", p) + } + + // VerifyClaims returns the rich claim set incl. email/name. + rawTok := iss.mint(t, valid) + c, err := o.VerifyClaims(rawTok) + if err != nil { + t.Fatalf("VerifyClaims rejected valid token: %v", err) + } + if c.Subject != "user-1" || c.Email != "user-1@example.com" || c.Name != "User One" || len(c.Scopes) != 2 { + t.Fatalf("unexpected claims: %+v", c) + } + + // aud as array form also accepted + arr := map[string]any{ + "iss": iss.srv.URL, "sub": "user-1", "aud": []string{"other", "my-client"}, + "exp": time.Now().Add(time.Hour).Unix(), + } + if _, err := o.Authenticate(bearerReq(iss.mint(t, arr))); err != nil { + t.Fatalf("array aud rejected: %v", err) + } + + cases := map[string]map[string]any{ + "expired": {"iss": iss.srv.URL, "sub": "u", "aud": "my-client", "exp": time.Now().Add(-time.Hour).Unix()}, + "wrong issuer": {"iss": "https://evil.example", "sub": "u", "aud": "my-client", "exp": time.Now().Add(time.Hour).Unix()}, + "wrong audience": {"iss": iss.srv.URL, "sub": "u", "aud": "not-me", "exp": time.Now().Add(time.Hour).Unix()}, + } + for name, c := range cases { + if _, err := o.Authenticate(bearerReq(iss.mint(t, c))); err == nil { + t.Errorf("%s token accepted", name) + } + if _, err := o.VerifyClaims(iss.mint(t, c)); err == nil { + t.Errorf("VerifyClaims accepted %s token", name) + } + } + + // tampered payload must fail signature verification + tok := iss.mint(t, valid) + tampered := fmt.Sprintf("%s.%s.%s", + tok[:len(tok)/3], + base64.RawURLEncoding.EncodeToString([]byte(`{"iss":"x","sub":"hacker"}`)), + "AAAA") + if _, err := o.Authenticate(bearerReq(tampered)); err == nil { + t.Error("tampered token accepted") + } + if _, err := o.VerifyClaims(tampered); err == nil { + t.Error("VerifyClaims accepted tampered token") + } + + // no credentials → ErrNoCredentials so Chain falls through + if _, err := o.Authenticate(httptest.NewRequest("GET", "/", nil)); err == nil { + t.Error("missing bearer accepted") + } + if _, err := o.VerifyClaims(""); err == nil { + t.Error("VerifyClaims accepted empty token") + } +} + +// --- API keys --- + +func TestKeyStoreMintAuthenticateRevoke(t *testing.T) { + path := filepath.Join(t.TempDir(), "apikeys.json") + ks, err := OpenKeyStore(path) + if err != nil { + t.Fatal(err) + } + key, rec, err := ks.Mint("ci", []string{"nodes:read"}) + if err != nil { + t.Fatal(err) + } + + a := APIKeyAuth{Keys: ks} + r := httptest.NewRequest("GET", "/api/nodes", nil) + r.Header.Set("X-API-Key", key) + p, err := a.Authenticate(r) + if err != nil { + t.Fatalf("minted key rejected: %v", err) + } + if p.Subject != rec.ID || p.Kind != KindAPIKey || !p.HasScope("nodes:read") { + t.Fatalf("unexpected principal: %+v", p) + } + + // persists across reopen + ks2, err := OpenKeyStore(path) + if err != nil { + t.Fatal(err) + } + if _, err := (APIKeyAuth{Keys: ks2}).Authenticate(r); err != nil { + t.Fatalf("key lost after reopen: %v", err) + } + + // invalid key rejected + bad := httptest.NewRequest("GET", "/", nil) + bad.Header.Set("X-API-Key", "gck_not_a_real_key") + if _, err := a.Authenticate(bad); err == nil { + t.Error("invalid key accepted") + } + + // revoked key rejected + if err := ks.Revoke(rec.ID); err != nil { + t.Fatal(err) + } + if _, err := a.Authenticate(r); err == nil { + t.Error("revoked key accepted") + } + if err := ks.Revoke(rec.ID); err == nil { + t.Error("revoking twice should fail") + } +} + +func TestChainBearerThenAPIKey(t *testing.T) { + iss := newFakeIssuer(t) + ks, err := OpenKeyStore(filepath.Join(t.TempDir(), "k.json")) + if err != nil { + t.Fatal(err) + } + key, _, err := ks.Mint("integration", nil) + if err != nil { + t.Fatal(err) + } + chain := Chain{NewOIDC(iss.srv.URL, ""), APIKeyAuth{Keys: ks}} + + r := httptest.NewRequest("GET", "/", nil) + r.Header.Set("X-API-Key", key) + p, err := chain.Authenticate(r) + if err != nil { + t.Fatalf("api key via chain rejected: %v", err) + } + if p.Kind != KindAPIKey { + t.Fatalf("expected apikey principal, got %+v", p) + } + + tok := iss.mint(t, map[string]any{"iss": iss.srv.URL, "sub": "u1", "exp": time.Now().Add(time.Hour).Unix()}) + p, err = chain.Authenticate(bearerReq(tok)) + if err != nil { + t.Fatalf("bearer via chain rejected: %v", err) + } + if p.Kind != KindOIDC { + t.Fatalf("expected oidc principal, got %+v", p) + } + + if _, err := chain.Authenticate(httptest.NewRequest("GET", "/", nil)); err == nil { + t.Error("credential-less request accepted by chain") + } +} + +// --- middleware --- + +func TestBearerMiddleware(t *testing.T) { + iss := newFakeIssuer(t) + tok := iss.mint(t, map[string]any{ + "iss": iss.srv.URL, "sub": "u", "aud": "client", + "exp": time.Now().Add(time.Hour).Unix(), + }) + a := NewOIDC(iss.srv.URL, "client") + hits := 0 + handler := BearerMiddleware(a, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + hits++ + w.WriteHeader(http.StatusOK) + })) + + // valid token → passes through + w := httptest.NewRecorder() + handler.ServeHTTP(w, bearerReq(tok)) + if w.Code != http.StatusOK || hits != 1 { + t.Fatalf("valid token: code=%d hits=%d", w.Code, hits) + } + + // missing token → 401 + w = httptest.NewRecorder() + handler.ServeHTTP(w, httptest.NewRequest("GET", "/", nil)) + if w.Code != http.StatusUnauthorized || hits != 1 { + t.Fatalf("missing token: code=%d hits=%d", w.Code, hits) + } +} + +func TestWriteMiddlewarePassesReads(t *testing.T) { + a := NewOIDC("https://issuer.example", "") + hits := 0 + handler := WriteMiddleware(a, http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { hits++ })) + + for _, m := range []string{http.MethodGet, http.MethodHead, http.MethodOptions} { + w := httptest.NewRecorder() + handler.ServeHTTP(w, httptest.NewRequest(m, "/", nil)) + if w.Code != http.StatusOK { + t.Errorf("%s: code=%d, want 200", m, w.Code) + } + } +} diff --git a/auth/middleware.go b/auth/middleware.go new file mode 100644 index 0000000..0fa2f19 --- /dev/null +++ b/auth/middleware.go @@ -0,0 +1,35 @@ +package auth + +import "net/http" + +// BearerMiddleware wraps next, requiring a valid bearer credential from a. +// On success the request is forwarded unchanged; on failure a 401 is +// returned. +func BearerMiddleware(a Authenticator, next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if _, err := a.Authenticate(r); err != nil { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + next.ServeHTTP(w, r) + }) +} + +// WriteMiddleware wraps next, requiring a valid bearer credential from a +// only for mutating HTTP methods (POST, PUT, PATCH, DELETE). Safe methods +// (GET, HEAD, OPTIONS) pass through unauthenticated — suitable for +// services that serve public reads but gate writes behind auth. +func WriteMiddleware(a Authenticator, next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.Method { + case http.MethodGet, http.MethodHead, http.MethodOptions: + next.ServeHTTP(w, r) + default: + if _, err := a.Authenticate(r); err != nil { + http.Error(w, "unauthorized", http.StatusUnauthorized) + return + } + next.ServeHTTP(w, r) + } + }) +} diff --git a/auth/oidc.go b/auth/oidc.go new file mode 100644 index 0000000..251c610 --- /dev/null +++ b/auth/oidc.go @@ -0,0 +1,337 @@ +package auth + +import ( + "crypto" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rsa" + "crypto/sha256" + "encoding/base64" + "encoding/json" + "errors" + "fmt" + "math/big" + "net/http" + "slices" + "strings" + "sync" + "time" +) + +// OIDC validates `Authorization: Bearer ` against any OIDC issuer: it +// fetches `/.well-known/openid-configuration` once, caches the JWKS, +// and verifies signature (RS256/ES256), iss, aud and exp/nbf. Zitadel is the +// reference issuer but nothing here is provider-specific. +// +// Note: the issuer must mint JWT access tokens (for Zitadel, set the +// application's access-token type to JWT); opaque tokens require +// introspection, which this deliberately does not implement. +type OIDC struct { + issuer string + audience string // optional: when set, must appear in the token's aud + client *http.Client + + mu sync.Mutex + jwksURI string + keys map[string]crypto.PublicKey // kid → verification key + lastRefresh time.Time + + now func() time.Time // injectable clock for tests +} + +var _ Authenticator = (*OIDC)(nil) + +// NewOIDC builds an OIDC authenticator for the given issuer URL. audience +// is the expected `aud` claim (typically the OAuth client id); empty skips +// the audience check. +func NewOIDC(issuer, audience string) *OIDC { + return &OIDC{ + issuer: strings.TrimSuffix(issuer, "/"), + audience: audience, + client: &http.Client{Timeout: 10 * time.Second}, + now: time.Now, + } +} + +// Authenticate validates the bearer JWT on r and returns the caller as a +// minimal Principal. For callers that need the full parsed claim set +// (Email/Name/scope list) — e.g. admin handlers rendering who is logged in — +// see VerifyClaims. +func (o *OIDC) Authenticate(r *http.Request) (*Principal, error) { + h := r.Header.Get("Authorization") + token, ok := strings.CutPrefix(h, "Bearer ") + if !ok || token == "" { + return nil, ErrNoCredentials + } + c, err := o.VerifyClaims(token) + if err != nil { + return nil, err + } + return &Principal{Subject: c.Subject, Scopes: c.Scopes, Kind: KindOIDC}, nil +} + +// Claims is the parsed, verified claim set returned by VerifyClaims. Only +// the subset services commonly need is exposed; the raw claim JSON is +// available via the original token, which callers can re-decode if they +// need something custom. +type Claims struct { + Issuer string `json:"iss"` + Subject string `json:"sub"` + Email string `json:"email,omitempty"` + Name string `json:"name,omitempty"` + Audience Audience `json:"aud"` + Expires int64 `json:"exp"` + NotBefore int64 `json:"nbf"` + Scopes []string `json:"scope,omitempty"` // parsed from "scope" space-separated claim +} + +// Audience unmarshals the `aud` claim, which RFC 7519 allows to be either a +// string or an array of strings (some providers emit one, some the other; +// Zitadel emits an array). +type Audience []string + +func (a *Audience) UnmarshalJSON(b []byte) error { + var s string + if err := json.Unmarshal(b, &s); err == nil { + *a = Audience{s} + return nil + } + var ss []string + if err := json.Unmarshal(b, &ss); err != nil { + return err + } + *a = Audience(ss) + return nil +} + +// contains reports whether want appears in the audience list. +func (a Audience) contains(want string) bool { + return slices.Contains(a, want) +} + +// clockSkew is the tolerance applied when checking expiry/notBefore so a +// small clock drift between issuer and verifier doesn't reject a fresh +// token. +const clockSkew = 30 * time.Second + +// VerifyClaims verifies a raw JWT (no "Bearer " prefix) end-to-end and +// returns the parsed claim set. Callers that aren't using the standard +// Authorization header (e.g. an admin UI reading a session cookie) can +// pass the cookie value directly. The token is verified for issuer, +// audience (if configured), expiry/notBefore, and signature via the JWKS +// cached against the configured issuer. +func (o *OIDC) VerifyClaims(token string) (*Claims, error) { + parts := strings.Split(token, ".") + if len(parts) != 3 { + return nil, errors.New("oidc: malformed jwt") + } + headerJSON, err := base64.RawURLEncoding.DecodeString(parts[0]) + if err != nil { + return nil, fmt.Errorf("oidc: bad jwt header: %w", err) + } + var header struct { + Alg string `json:"alg"` + Kid string `json:"kid"` + } + if err := json.Unmarshal(headerJSON, &header); err != nil { + return nil, fmt.Errorf("oidc: bad jwt header: %w", err) + } + payloadJSON, err := base64.RawURLEncoding.DecodeString(parts[1]) + if err != nil { + return nil, fmt.Errorf("oidc: bad jwt payload: %w", err) + } + sig, err := base64.RawURLEncoding.DecodeString(parts[2]) + if err != nil { + return nil, fmt.Errorf("oidc: bad jwt signature: %w", err) + } + + key, err := o.key(header.Kid) + if err != nil { + return nil, err + } + digest := sha256.Sum256([]byte(parts[0] + "." + parts[1])) + switch header.Alg { + case "RS256": + pub, ok := key.(*rsa.PublicKey) + if !ok { + return nil, errors.New("oidc: key/alg mismatch") + } + if err := rsa.VerifyPKCS1v15(pub, crypto.SHA256, digest[:], sig); err != nil { + return nil, errors.New("oidc: invalid signature") + } + case "ES256": + pub, ok := key.(*ecdsa.PublicKey) + if !ok { + return nil, errors.New("oidc: key/alg mismatch") + } + if len(sig) != 64 { + return nil, errors.New("oidc: invalid signature") + } + rr := new(big.Int).SetBytes(sig[:32]) + ss := new(big.Int).SetBytes(sig[32:]) + if !ecdsa.Verify(pub, digest[:], rr, ss) { + return nil, errors.New("oidc: invalid signature") + } + default: + return nil, fmt.Errorf("oidc: unsupported alg %q", header.Alg) + } + + // Decode into a wide struct so Email/Name are picked up regardless of + // claim shape (some providers omit them, some populate them). + var raw struct { + Issuer string `json:"iss"` + Subject string `json:"sub"` + Email string `json:"email"` + Name string `json:"name"` + Audience Audience `json:"aud"` + Expires int64 `json:"exp"` + NotBefore int64 `json:"nbf"` + Scope string `json:"scope"` + } + if err := json.Unmarshal(payloadJSON, &raw); err != nil { + return nil, fmt.Errorf("oidc: bad claims: %w", err) + } + now := o.now() + if strings.TrimSuffix(raw.Issuer, "/") != o.issuer { + return nil, errors.New("oidc: wrong issuer") + } + if raw.Expires != 0 && now.After(time.Unix(raw.Expires, 0).Add(clockSkew)) { + return nil, errors.New("oidc: token expired") + } + if raw.NotBefore != 0 && now.Add(clockSkew).Before(time.Unix(raw.NotBefore, 0)) { + return nil, errors.New("oidc: token not yet valid") + } + if o.audience != "" && !raw.Audience.contains(o.audience) { + return nil, errors.New("oidc: wrong audience") + } + + var scopes []string + if raw.Scope != "" { + scopes = strings.Fields(raw.Scope) + } + return &Claims{ + Issuer: raw.Issuer, + Subject: raw.Subject, + Email: raw.Email, + Name: raw.Name, + Audience: raw.Audience, + Expires: raw.Expires, + NotBefore: raw.NotBefore, + Scopes: scopes, + }, nil +} + +// key returns the verification key for kid, fetching/refreshing the JWKS +// when the kid is unknown (key rotation), rate-limited to once a minute. +func (o *OIDC) key(kid string) (crypto.PublicKey, error) { + o.mu.Lock() + defer o.mu.Unlock() + if k, ok := o.keys[kid]; ok { + return k, nil + } + if time.Since(o.lastRefresh) < time.Minute { + return nil, fmt.Errorf("oidc: unknown key id %q", kid) + } + if err := o.refreshLocked(); err != nil { + return nil, err + } + if k, ok := o.keys[kid]; ok { + return k, nil + } + return nil, fmt.Errorf("oidc: unknown key id %q", kid) +} + +// refreshLocked (re)fetches the discovery document (first time) and the +// JWKS. Callers must hold o.mu. +func (o *OIDC) refreshLocked() error { + o.lastRefresh = o.now() + if o.jwksURI == "" { + var disc struct { + JWKSURI string `json:"jwks_uri"` + } + if err := o.getJSON(o.issuer+"/.well-known/openid-configuration", &disc); err != nil { + return fmt.Errorf("oidc: discovery: %w", err) + } + if disc.JWKSURI == "" { + return errors.New("oidc: discovery document has no jwks_uri") + } + o.jwksURI = disc.JWKSURI + } + var set struct { + Keys []jwk `json:"keys"` + } + if err := o.getJSON(o.jwksURI, &set); err != nil { + return fmt.Errorf("oidc: jwks: %w", err) + } + keys := make(map[string]crypto.PublicKey, len(set.Keys)) + for _, k := range set.Keys { + pub, err := k.publicKey() + if err != nil { + continue // skip unsupported key types (e.g. enc-use keys) + } + keys[k.Kid] = pub + } + o.keys = keys + return nil +} + +func (o *OIDC) getJSON(url string, v any) error { + res, err := o.client.Get(url) + if err != nil { + return err + } + defer res.Body.Close() + if res.StatusCode != http.StatusOK { + return fmt.Errorf("GET %s: %s", url, res.Status) + } + return json.NewDecoder(res.Body).Decode(v) +} + +// jwk is the subset of RFC 7517 needed to build RSA and P-256 keys. +type jwk struct { + Kty string `json:"kty"` + Kid string `json:"kid"` + Use string `json:"use"` + N string `json:"n"` + E string `json:"e"` + Crv string `json:"crv"` + X string `json:"x"` + Y string `json:"y"` +} + +func (k jwk) publicKey() (crypto.PublicKey, error) { + switch k.Kty { + case "RSA": + n, err := base64.RawURLEncoding.DecodeString(k.N) + if err != nil { + return nil, err + } + eBytes, err := base64.RawURLEncoding.DecodeString(k.E) + if err != nil { + return nil, err + } + return &rsa.PublicKey{ + N: new(big.Int).SetBytes(n), + E: int(new(big.Int).SetBytes(eBytes).Int64()), + }, nil + case "EC": + if k.Crv != "P-256" { + return nil, fmt.Errorf("unsupported curve %q", k.Crv) + } + x, err := base64.RawURLEncoding.DecodeString(k.X) + if err != nil { + return nil, err + } + y, err := base64.RawURLEncoding.DecodeString(k.Y) + if err != nil { + return nil, err + } + return &ecdsa.PublicKey{ + Curve: elliptic.P256(), + X: new(big.Int).SetBytes(x), + Y: new(big.Int).SetBytes(y), + }, nil + default: + return nil, fmt.Errorf("unsupported key type %q", k.Kty) + } +} diff --git a/catalog-emit b/catalog-emit deleted file mode 100755 index 89dc757..0000000 Binary files a/catalog-emit and /dev/null differ diff --git a/catalog/consumer.go b/catalog/consumer.go index 54348e0..044a92a 100644 --- a/catalog/consumer.go +++ b/catalog/consumer.go @@ -1,330 +1,99 @@ package catalog -// ProjectionStore — the consumer half of the projection seam (see -// docs/catalog-projection.md). A commerce service (cart, checkout, order, UCP) -// feeds it the catalog.projection_published frames it receives off the bus and -// reads catalog facts back by SKU. The store owns a LOCAL read-only projection: +// Projection seam — catalog binding over the generic platform/projection +// package. The value-agnostic machinery (snapshot .bin format, mmap reader, +// race-free SnapshotHolder, the begin→chunk*→end consumer state machine with +// epoch gating + delta overlay) lives in platform/projection. This file is the +// thin catalog-specific binding: it fixes the WIRE type to catalog.Projection +// and supplies the decoder for the catalog wire schema, so cart / checkout / +// order / catalog-emit keep their existing catalog.* API unchanged. // -// - A full snapshot stream (begin → chunk* → end, one MetaEpoch) is assembled -// into the store's OWN .bin (SnapshotWriter), atomically swapped in as the -// mmap base (SnapshotHolder). The base is the cold-start substrate — it -// survives a restart (reopened from disk), so the consumer serves reads -// immediately without waiting for a bus replay. -// - Per-batch deltas of the SAME epoch are applied to an in-memory overlay on -// top of the base (overlay wins; a delete leaves a tombstone that shadows the -// base and the fallback). The next snapshot folds the overlay into a new base. -// -// Read order on Get: overlay → mmap base → optional caller fallback (e.g. an -// HTTP fetch from the reader) — last-resort only. The access pattern is point -// lookup by SKU; nothing enumerates the catalog. -// -// No shared volume: each consumer builds its own base from the bus. Concurrency: -// HandleFrame is single-writer (drive it from one delivery goroutine); Get is -// safe from many readers. +// See docs/catalog-projection.md for the consumer contract and +// docs/plan-projectionstore-generalization.md for why the split exists. import ( - "fmt" - "log" - "os" - "path/filepath" - "strconv" - "strings" - "sync" + "time" + + "git.k6n.net/mats/platform/projection" ) -// ProjectionFallback resolves an SKU the local store doesn't have yet (cold -// start before the first snapshot, or an item added since the last base). It -// returns a raw Projection — the store applies the SAME Transform to it, so a -// fallback hit is shaped identically to bus-fed data. Return (_, false) to -// report a genuine miss. -type ProjectionFallback func(sku string) (Projection, bool) +// Generic machinery, re-exported under their historical catalog names. These +// are type aliases — catalog.SnapshotHolder[T] IS projection.SnapshotHolder[T], +// so values flow between the two packages without conversion. +type ( + SnapshotWriter[T any] = projection.SnapshotWriter[T] + SnapshotReader[T any] = projection.SnapshotReader[T] + SnapshotHolder[T any] = projection.SnapshotHolder[T] -// ProjectionTransform projects the wire Projection down to the consumer's own -// stored type T. It is MANDATORY: it is the one place a consumer declares which -// catalog facts it actually needs (and any derivation, e.g. tax_class → rate), -// so its local .bin holds only that — smaller, and read-ready with no per-Get -// conversion. Every value entering the store — snapshot, delta, and fallback — -// passes through it exactly once. -type ProjectionTransform[T any] func(Projection) T + // ProjectionStore is a consumer's local, cold-start-ready cache of T, fed the + // catalog.projection_published frames off the bus. The wire type is fixed to + // catalog.Projection; T is the consumer's projected shape. + ProjectionStore[T any] = projection.Store[Projection, T] -// ProjectionStore is a consumer's local, cold-start-ready projection cache of T. -type ProjectionStore[T any] struct { - dir, name string - transform ProjectionTransform[T] - fallback ProjectionFallback - holder *SnapshotHolder[T] // mmap base; atomic-swap + drain-grace + // ProjectionTransform projects the wire Projection down to the consumer's + // stored type T (the one place a consumer declares which catalog facts it + // needs). MANDATORY. + ProjectionTransform[T any] = projection.Transform[Projection, T] - mu sync.RWMutex - epoch int64 // epoch of the live base - overlay map[string]overlayEntry[T] // same-epoch deltas layered on the base + // ProjectionFallback resolves a SKU the local store doesn't have yet. May be nil. + ProjectionFallback = projection.Fallback[Projection] +) - // in-progress snapshot assembly (one epoch at a time) - building *SnapshotWriter[T] - buildingEpoch int64 - buildingSeen int +// DefaultDrainGrace is the SnapshotHolder default drain period (re-exported). +const DefaultDrainGrace = projection.DefaultDrainGrace + +// Snapshot file errors, re-exported so callers can errors.Is against the +// catalog.* names. Identity is preserved (these ARE the projection package's +// error values). +var ( + ErrSnapshotMagic = projection.ErrSnapshotMagic + ErrSnapshotVersion = projection.ErrSnapshotVersion + ErrSnapshotTruncated = projection.ErrSnapshotTruncated + ErrSnapshotBadBlob = projection.ErrSnapshotBadBlob + ErrSnapshotOverflow = projection.ErrSnapshotOverflow + ErrSKUTooLong = projection.ErrSKUTooLong + ErrTenantTooLong = projection.ErrTenantTooLong +) + +// NewSnapshotWriter returns an empty snapshot writer for value type T. +func NewSnapshotWriter[T any]() *SnapshotWriter[T] { return projection.NewSnapshotWriter[T]() } + +// NewSnapshotReader wraps an already-loaded []byte as a reader of T. +func NewSnapshotReader[T any](data []byte) (*SnapshotReader[T], error) { + return projection.NewSnapshotReader[T](data) } -type overlayEntry[T any] struct { - val T - deleted bool +// ReadSnapshotFile opens + parses a snapshot file as a reader of T (mmap on Unix). +func ReadSnapshotFile[T any](path string) (*SnapshotReader[T], error) { + return projection.ReadSnapshotFile[T](path) } -// NewProjectionStore opens a store of T backed by dir/name (its own .bin). -// transform is REQUIRED — it defines the consumer's projection shape (see -// ProjectionTransform). If a base from a previous run is present it is reopened -// immediately (cold-start ready); its epoch is read from the sidecar so -// same-epoch deltas can resume. fallback may be nil. Close() when done. +// NewSnapshotHolder returns a holder of T with the given drain grace. +func NewSnapshotHolder[T any](grace time.Duration) *SnapshotHolder[T] { + return projection.NewSnapshotHolder[T](grace) +} + +// NewProjectionStore opens a catalog ProjectionStore backed by dir/name (its own +// .bin). transform is REQUIRED (declares the stored shape); fallback may be nil. +// If a base from a previous run is present it is reopened immediately +// (cold-start ready). Close() when done. func NewProjectionStore[T any](dir, name string, transform ProjectionTransform[T], fallback ProjectionFallback) (*ProjectionStore[T], error) { - if dir == "" || name == "" { - return nil, fmt.Errorf("catalog: ProjectionStore needs dir and name") - } - if transform == nil { - return nil, fmt.Errorf("catalog: ProjectionStore needs a transform (declare which catalog facts this consumer stores)") - } - s := &ProjectionStore[T]{ - dir: dir, - name: name, - transform: transform, - fallback: fallback, - holder: NewSnapshotHolder[T](0), - overlay: make(map[string]overlayEntry[T]), - } - path := filepath.Join(dir, name) - if err := s.holder.Open(path); err == nil { - s.epoch = readEpochSidecar(path) - log.Printf("catalog: ProjectionStore reopened base %s (epoch=%d, %d entries)", path, s.epoch, s.holder.Len()) - } else if !os.IsNotExist(err) { - // A corrupt/partial base on disk shouldn't crash boot — start empty and - // wait for the next snapshot; reads fall back meanwhile. - log.Printf("catalog: ProjectionStore ignoring unreadable base %s: %v", path, err) - } - return s, nil + return projection.NewStore[Projection, T](dir, name, decodeProjectionUpdates, transform, fallback) } -// HandleFrame applies one catalog.projection_published event (snapshot frame or -// delta), routed by its Meta. Drive this from a single delivery goroutine. A -// returned error means the payload was undecodable (the caller may NACK); a -// frame that simply doesn't apply (stale epoch, wrong assembly state) returns -// nil so the delivery is acked. -func (s *ProjectionStore[T]) HandleFrame(meta map[string]string, payload []byte) error { - epoch, _ := strconv.ParseInt(meta[MetaEpoch], 10, 64) - switch meta[MetaSnapshotPhase] { - case SnapshotBegin: - return s.beginSnapshot(epoch, meta[MetaTenant]) - case SnapshotChunk: - return s.chunkSnapshot(epoch, payload) - case SnapshotEnd: - return s.endSnapshot(epoch, meta) - case "": - return s.applyDelta(epoch, payload) - default: - return nil // unknown phase — ignore forward-compatibly - } -} - -func (s *ProjectionStore[T]) beginSnapshot(epoch int64, tenant string) error { - s.mu.Lock() - defer s.mu.Unlock() - if epoch <= s.epoch { - return nil // we already hold this base or newer - } - // Start (or restart) assembly for this epoch. A new begin supersedes any - // half-built snapshot (e.g. a producer that aborted mid-stream). - s.building = NewSnapshotWriter[T]().WithTenant(tenant) - s.buildingEpoch = epoch - s.buildingSeen = 0 - return nil -} - -func (s *ProjectionStore[T]) chunkSnapshot(epoch int64, payload []byte) error { - ups, err := DecodeUpdates(payload) +// decodeProjectionUpdates bridges the catalog wire schema ([]ProjectionUpdate — +// a flat Projection plus a `deleted` flag) to the generic projection.Update the +// Store works with. The index key is the Projection's SKU. This is the one +// catalog-specific seam the generic Store can't know: the on-the-wire payload +// shape is a domain decision. +func decodeProjectionUpdates(payload []byte) ([]projection.Update[Projection], error) { + raw, err := DecodeUpdates(payload) if err != nil { - return fmt.Errorf("catalog: decode snapshot chunk: %w", err) + return nil, err } - s.mu.Lock() - defer s.mu.Unlock() - if s.building == nil || epoch != s.buildingEpoch { - return nil // not assembling this epoch + out := make([]projection.Update[Projection], len(raw)) + for i, u := range raw { + out[i] = projection.Update[Projection]{Key: u.SKU, Value: u.Projection, Deleted: u.Deleted} } - for _, u := range ups { - if u.Deleted || u.SKU == "" { - continue // a snapshot carries live items only - } - // Transform applied here — the base stores the consumer's shape, not the - // full Projection. - if err := s.building.Append(u.SKU, s.transform(u.Projection)); err != nil { - log.Printf("catalog: ProjectionStore skip %q in snapshot: %v", u.SKU, err) - continue - } - s.buildingSeen++ - } - return nil -} - -func (s *ProjectionStore[T]) endSnapshot(epoch int64, meta map[string]string) error { - s.mu.Lock() - defer s.mu.Unlock() - if s.building == nil || epoch != s.buildingEpoch { - return nil - } - w := s.building - s.building = nil - - // Completeness guard: if begin/end advertised a count and we appended fewer - // (a lost chunk), DON'T swap — keep the old base and wait for the next - // snapshot. Serving an incomplete catalog silently is worse than stale. - if cs := meta[MetaCount]; cs != "" { - if want, perr := strconv.Atoi(cs); perr == nil && want != s.buildingSeen { - log.Printf("catalog: ProjectionStore snapshot epoch=%d incomplete (%d/%d) — discarding, keeping base epoch=%d", - epoch, s.buildingSeen, want, s.epoch) - return nil - } - } - - path, err := w.Write(s.dir, s.name) - if err != nil { - return fmt.Errorf("catalog: write snapshot base: %w", err) - } - // Persist the epoch sidecar (atomic) BEFORE swapping the in-memory reader, so - // a reopened store's epoch always matches the .bin already on disk (w.Write - // renamed the new file into place above). Writing it after Reload would leave - // a crash window where the new .bin carries a stale/absent epoch — the store - // would then mislabel its base and drop the current epoch's deltas on restart. - writeEpochSidecar(path, epoch) - if err := s.holder.Reload(path); err != nil { - return fmt.Errorf("catalog: reload snapshot base: %w", err) - } - s.epoch = epoch - s.overlay = make(map[string]overlayEntry[T]) // base now includes everything ≤ epoch - log.Printf("catalog: ProjectionStore snapshot applied — base swapped epoch=%d (%d entries)", epoch, s.buildingSeen) - return nil -} - -func (s *ProjectionStore[T]) applyDelta(epoch int64, payload []byte) error { - ups, err := DecodeUpdates(payload) - if err != nil { - return fmt.Errorf("catalog: decode delta: %w", err) - } - s.mu.Lock() - defer s.mu.Unlock() - switch { - case s.building != nil: - // Under FIFO delivery a delta won't interleave a snapshot stream; if it - // does, the snapshot is authoritative for its epoch, so drop the delta. - return nil - case epoch < s.epoch: - return nil // stale — predates our base - case epoch > s.epoch: - // CONTRACT: a future-epoch delta is DROPPED, not buffered. Under FIFO - // per-queue delivery begin(N) always precedes delta(N), so this only - // happens after a reconnect that missed the whole snapshot — which a - // buffer couldn't reassemble anyway. Recovery is the next snapshot of - // this epoch; the consumer's fallback covers reads until then. - return nil - } - var upserts, deletes int - for _, u := range ups { - if u.SKU == "" { - continue - } - if u.Deleted { - s.overlay[u.SKU] = overlayEntry[T]{deleted: true} - deletes++ - continue - } - // Transform applied here too — overlay holds the consumer's shape. - s.overlay[u.SKU] = overlayEntry[T]{val: s.transform(u.Projection)} - upserts++ - } - if upserts+deletes > 0 { - log.Printf("catalog: ProjectionStore delta applied epoch=%d (+%d/-%d, overlay=%d)", - epoch, upserts, deletes, len(s.overlay)) - } - return nil -} - -// Get resolves a value by SKU: overlay (incl. tombstone) → mmap base → -// fallback. A tombstoned SKU returns (zero, false) and never reaches the -// fallback — the catalog explicitly removed it. A fallback hit is run through -// the same Transform, so it's shaped like everything else. -func (s *ProjectionStore[T]) Get(sku string) (T, bool) { - var zero T - s.mu.RLock() - e, ok := s.overlay[sku] - s.mu.RUnlock() - if ok { - if e.deleted { - return zero, false - } - return e.val, true - } - if v, ok := s.holder.Get(sku); ok { - return v, true - } - if s.fallback != nil { - if p, ok := s.fallback(sku); ok { - return s.transform(p), true - } - } - return zero, false -} - -// IsDeleted reports whether the overlay holds a tombstone for sku — a delete -// seen on the bus since the current base. Lets a caller skip a downstream fetch -// for an SKU the catalog just removed. After a snapshot folds the delete into a -// new base the SKU is simply absent (IsDeleted → false); "absent" is then the -// caller's fetch/miss path, which a deleted item resolves as a downstream 404. -func (s *ProjectionStore[T]) IsDeleted(sku string) bool { - s.mu.RLock() - defer s.mu.RUnlock() - e, ok := s.overlay[sku] - return ok && e.deleted -} - -// Epoch returns the epoch of the current base (0 if none yet). -func (s *ProjectionStore[T]) Epoch() int64 { - s.mu.RLock() - defer s.mu.RUnlock() - return s.epoch -} - -// Stats returns observability counters (base size, overlay size, epoch). -func (s *ProjectionStore[T]) Stats() (epoch int64, baseCount, overlayCount int) { - s.mu.RLock() - defer s.mu.RUnlock() - return s.epoch, s.holder.Len(), len(s.overlay) -} - -// Close releases the mmap base. -func (s *ProjectionStore[T]) Close() error { return s.holder.Close() } - -// ── epoch sidecar ──────────────────────────────────────────────────────────── -// The .bin header has no room for the epoch, so it's persisted next to it as -// ".epoch" — read on boot so a reopened base can resume same-epoch deltas. - -func epochSidecarPath(binPath string) string { return binPath + ".epoch" } - -// writeEpochSidecar writes the epoch atomically (temp+rename) so a crash mid-write -// can't leave a torn file that parses as a bogus epoch — a reader either sees the -// old epoch or the new one, never garbage. -func writeEpochSidecar(binPath string, epoch int64) { - p := epochSidecarPath(binPath) - tmp := p + ".tmp" - if err := os.WriteFile(tmp, []byte(strconv.FormatInt(epoch, 10)), 0o644); err != nil { - log.Printf("catalog: ProjectionStore epoch sidecar write: %v", err) - return - } - if err := os.Rename(tmp, p); err != nil { - log.Printf("catalog: ProjectionStore epoch sidecar rename: %v", err) - _ = os.Remove(tmp) - } -} - -func readEpochSidecar(binPath string) int64 { - b, err := os.ReadFile(epochSidecarPath(binPath)) - if err != nil { - return 0 - } - n, _ := strconv.ParseInt(strings.TrimSpace(string(b)), 10, 64) - return n + return out, nil } diff --git a/catalog/projection.go b/catalog/projection.go index 2d14794..ee22087 100644 --- a/catalog/projection.go +++ b/catalog/projection.go @@ -36,6 +36,7 @@ import ( "encoding/json" "git.k6n.net/mats/platform/money" + "git.k6n.net/mats/platform/projection" ) // Projection is the published, versioned catalog entry that commerce modules @@ -82,10 +83,10 @@ import ( // the JSON token. The on-disk snapshot binary header/record layout is // unchanged \u2014 only the per-blob JSON payload gains a key. type Projection struct { - ItemID uint32 `json:"item_id,omitempty"` - ID string `json:"id"` - SKU string `json:"sku"` - Title string `json:"title,omitempty"` + ItemID uint32 `json:"item_id,omitempty"` + ID string `json:"id"` + SKU string `json:"sku"` + Title string `json:"title,omitempty"` PriceIncVat money.Cents `json:"price_inc_vat"` Currency string `json:"currency,omitempty"` // TaxClass uses a string identifier (see platform/tax class constants). @@ -154,7 +155,9 @@ func Decode(raw []byte) (Projection, error) { // catalog.projection_published carries two channels distinguished only by // event.Event.Meta. The producer (slask-finder writeradmin) sets these keys and // the consumer (ProjectionStore, consumer.go) reads them — one definition so the -// two never drift. See docs/catalog-projection.md. +// two never drift. The framing itself is domain-neutral and lives in +// platform/projection; catalog re-exports the constants so the producer/consumer +// keep their catalog.* names. See docs/catalog-projection.md. // // - A FULL SNAPSHOT is a framed begin → chunk* → end stream, all tagged with one // MetaEpoch. begin/end carry MetaCount (total projections) for completeness @@ -162,13 +165,13 @@ func Decode(raw []byte) (Projection, error) { // - A DELTA has NO MetaSnapshotPhase key, and MetaEpoch names the snapshot base // it builds on (consumers ignore a delta older than the base they hold). const ( - MetaSnapshotPhase = "snapshot" // begin|chunk|end; absent ⇒ delta - MetaEpoch = "epoch" // int64 (Unix-ms), monotonic across rebuilds - MetaSeq = "seq" // chunk sequence within a snapshot - MetaCount = "count" // total projections (begin/end only) - MetaTenant = "tenant" // namespace tag + MetaSnapshotPhase = projection.MetaSnapshotPhase // begin|chunk|end; absent ⇒ delta + MetaEpoch = projection.MetaEpoch // int64 (Unix-ms), monotonic across rebuilds + MetaSeq = projection.MetaSeq // chunk sequence within a snapshot + MetaCount = projection.MetaCount // total projections (begin/end only) + MetaTenant = projection.MetaTenant // namespace tag - SnapshotBegin = "begin" - SnapshotChunk = "chunk" - SnapshotEnd = "end" + SnapshotBegin = projection.SnapshotBegin + SnapshotChunk = projection.SnapshotChunk + SnapshotEnd = projection.SnapshotEnd ) diff --git a/catalog/snapshot_helpers_test.go b/catalog/snapshot_helpers_test.go deleted file mode 100644 index ad5337a..0000000 --- a/catalog/snapshot_helpers_test.go +++ /dev/null @@ -1,28 +0,0 @@ -// Local helpers for snapshot_test.go — production code does not see them. -package catalog - -import ( - "os" -) - -func os2WriteFile(path string, data []byte, perm os.FileMode) error { - return os.WriteFile(path, data, perm) -} - -func os2Stat(path string) (os.FileInfo, error) { - return os.Stat(path) -} - -// appendP appends a Projection keyed by its own SKU — the producer/test pattern -// over the now-generic SnapshotWriter[Projection] (Append takes an explicit key). -func appendP(w *SnapshotWriter[Projection], p Projection) error { - return w.Append(p.SKU, p) -} - -func repeat32(s string, n int) string { - out := make([]byte, 0, n*len(s)) - for i := 0; i < n; i++ { - out = append(out, s...) - } - return string(out) -} diff --git a/cmd/catalog-emit/main.go b/cmd/catalog-emit/main.go index 0a4f768..bdf4bec 100644 --- a/cmd/catalog-emit/main.go +++ b/cmd/catalog-emit/main.go @@ -82,7 +82,7 @@ type TenantState struct { mu sync.RWMutex items map[string]catalog.Projection - holder *catalog.SnapshotHolder + holder *catalog.SnapshotHolder[catalog.Projection] path string // /.bin rebuildCount atomic.Uint64 @@ -92,7 +92,7 @@ type TenantState struct { // newTenantState wires a brand-new tenant. Caller is responsible for // calling rebuild() (or Open via the initial state pipeline) before any Get. -func newTenantState(tenant string, sch catalog.Schema, holder *catalog.SnapshotHolder, path string) *TenantState { +func newTenantState(tenant string, sch catalog.Schema, holder *catalog.SnapshotHolder[catalog.Projection], path string) *TenantState { return &TenantState{ tenant: tenant, schema: sch, @@ -148,9 +148,9 @@ func (ts *TenantState) rebuild(batch []catalog.ProjectionUpdate) error { // writeSnapshot emits snapshot to disk (atomic temp+rename) and reloads // the holder. Called WITHOUT holding ts.mu. func (ts *TenantState) writeSnapshot(snap []catalog.Projection) error { - w := catalog.NewSnapshotWriter().WithTenant(ts.tenant) + w := catalog.NewSnapshotWriter[catalog.Projection]().WithTenant(ts.tenant) for _, p := range snap { - if err := w.Append(p); err != nil { + if err := w.Append(p.SKU, p); err != nil { return fmt.Errorf("append %s: %w", p.SKU, err) } } @@ -260,7 +260,7 @@ func main() { sch.Tenant = t } path := filepath.Join(*outDir, t+".bin") - holder := catalog.NewSnapshotHolder(*drainGrace) + holder := catalog.NewSnapshotHolder[catalog.Projection](*drainGrace) states[t] = newTenantState(t, sch, holder, path) } // Register cleanup for both the defer path (graceful shutdown / SIGINT) diff --git a/go.mod b/go.mod index ee6c8b9..47b85e9 100644 --- a/go.mod +++ b/go.mod @@ -3,6 +3,6 @@ module git.k6n.net/mats/platform go 1.26.2 require ( - github.com/rabbitmq/amqp091-go v1.11.0 + github.com/rabbitmq/amqp091-go v1.12.0 golang.org/x/sys v0.46.0 ) diff --git a/go.sum b/go.sum index ffa9cde..c26aeed 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,4 @@ -github.com/rabbitmq/amqp091-go v1.11.0 h1:HxIctVm9Gid/Vtn706necmZ7Wj6pgGI2eqplRbEY8O8= -github.com/rabbitmq/amqp091-go v1.11.0/go.mod h1:Hy4jKW5kQART1u+JkDTF9YYOQUHXqMuhrgxOEeS7G4o= +github.com/rabbitmq/amqp091-go v1.12.0 h1:V0v14Iqfs+MwHWihJt/nGS5Ulu0vw572b2Co3mwunkI= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= golang.org/x/sys v0.46.0 h1:noSf2Fq6F8DBgS+LysIkx7rIExoNHJsxOAtPp4rthXw= diff --git a/inventory/reservation.go b/inventory/reservation.go new file mode 100644 index 0000000..20e2087 --- /dev/null +++ b/inventory/reservation.go @@ -0,0 +1,59 @@ +package inventory + +import ( + "context" + "time" +) + +// Value types for inventory reservation — cross-module vocabulary shared by +// cart, checkout, backoffice, and the inventory service. Moving them here +// means consumers depend on platform/inventory rather than on the Redis +// implementation package (go-redis-inventory). + +// SKU identifies a single stock-keeping unit. +type SKU string + +// LocationID identifies a geographic stock location (e.g. "se", "no"). +type LocationID string + +// CartID identifies a shopping cart — used as the reservation holder key. +type CartID string + +// InventoryReference identifies a specific SKU at a specific location. +type InventoryReference struct { + SKU SKU + LocationID LocationID +} + +// CartReserveRequest is what a caller provides to reserve stock for a cart. +type CartReserveRequest struct { + *InventoryReference + CartID CartID + Quantity uint32 + TTL time.Duration +} + +// ReservationPolicy is the cross-module seam for inventory reservation. +// Cart mutations call Reserve on add-to-cart / change-quantity and Release on +// item removal. Checkout calls Available to validate stock before a payment +// session, and Release on cancel / destroy. Every implementation behind this +// interface is free to choose its internal storage (Redis Lua, in-memory map, +// actor grain, etc.) — the consumer never needs to know. +// +// This is the "policy-first seam" from docs/ARCHITECTURE.md: +// commerce policy lives here; the fast, contended storage lives behind it. +type ReservationPolicy interface { + // Reserve holds qty units of sku at locationID for the caller's cart. + // Returns an error if insufficient stock is available. + Reserve(ctx context.Context, req CartReserveRequest) error + + // Release frees a prior reservation for the given cart at the given + // sku+location. Idempotent: releasing an already-released or expired + // reservation is a no-op. + Release(ctx context.Context, sku SKU, locationID LocationID, cartID CartID) error + + // Available returns the quantity that can still be reserved right now — + // total stock minus all active reservations. Used by checkout to validate + // that the cart's items are still reservable before opening a payment session. + Available(ctx context.Context, sku SKU, locationID LocationID) (int64, error) +} diff --git a/inventory/reservation_test.go b/inventory/reservation_test.go new file mode 100644 index 0000000..9b50eaf --- /dev/null +++ b/inventory/reservation_test.go @@ -0,0 +1,187 @@ +package inventory + +import ( + "context" + "sync" + "testing" + "time" +) + +// InMemoryReservationPolicy is a thread-safe in-memory implementation of +// ReservationPolicy for tests — no Redis needed. Backed by a map from +// SKU+location to available quantity. +type InMemoryReservationPolicy struct { + mu sync.Mutex + qty map[string]int64 // key: "sku|location" → available quantity + held map[string]int64 // key: "sku|location|cartID" → reserved quantity +} + +// NewInMemoryReservationPolicy creates a policy seeded with the given +// sku+location → available quantity entries. Each entry is a pair of +// (sku, location, qty); pass in threes. +func NewInMemoryReservationPolicy(seed ...any) *InMemoryReservationPolicy { + p := &InMemoryReservationPolicy{ + qty: make(map[string]int64), + held: make(map[string]int64), + } + for i := 0; i+2 < len(seed); i += 3 { + sku, ok := seed[i].(SKU) + if !ok { + continue + } + loc, ok := seed[i+1].(LocationID) + if !ok { + continue + } + qty, ok := seed[i+2].(int64) + if !ok { + continue + } + p.qty[key(sku, loc)] = qty + } + return p +} + +func key(sku SKU, loc LocationID) string { return string(sku) + "|" + string(loc) } +func heldKey(sku SKU, loc LocationID, c CartID) string { + return string(sku) + "|" + string(loc) + "|" + string(c) +} + +func (p *InMemoryReservationPolicy) Reserve(_ context.Context, req CartReserveRequest) error { + p.mu.Lock() + defer p.mu.Unlock() + + k := key(req.SKU, req.LocationID) + avail := p.qty[k] + hk := heldKey(req.SKU, req.LocationID, req.CartID) + if _, exists := p.held[hk]; exists { + // Already held — add to existing reservation + avail -= p.held[hk] + } + if avail < int64(req.Quantity) { + return ErrInsufficientStock + } + p.held[hk] += int64(req.Quantity) + return nil +} + +func (p *InMemoryReservationPolicy) Release(_ context.Context, sku SKU, locationID LocationID, cartID CartID) error { + p.mu.Lock() + defer p.mu.Unlock() + + delete(p.held, heldKey(sku, locationID, cartID)) + return nil +} + +func (p *InMemoryReservationPolicy) Available(_ context.Context, sku SKU, locationID LocationID) (int64, error) { + p.mu.Lock() + defer p.mu.Unlock() + + k := key(sku, locationID) + avail := p.qty[k] + // Subtract all held quantities for this sku+location + for hk, held := range p.held { + if len(hk) > len(k) && hk[:len(k)] == k && hk[len(k)] == '|' { + avail -= held + } + } + if avail < 0 { + avail = 0 + } + return avail, nil +} + +// Reset clears all reservations and resets available quantities. +func (p *InMemoryReservationPolicy) Reset() { + p.mu.Lock() + defer p.mu.Unlock() + p.held = make(map[string]int64) +} + +// ErrInsufficientStock is returned by Reserve when not enough stock is available. +var ErrInsufficientStock = &reservationError{"insufficient stock"} + +type reservationError struct{ msg string } + +func (e *reservationError) Error() string { return e.msg } + +// ── Tests ──────────────────────────────────────────────────────────────── + +func TestInMemoryReserveAndRelease(t *testing.T) { + ctx := context.Background() + p := NewInMemoryReservationPolicy( + SKU("ABC-1"), LocationID("se"), int64(10), + SKU("ABC-1"), LocationID("no"), int64(5), + ) + + // Reserve available stock + err := p.Reserve(ctx, CartReserveRequest{ + InventoryReference: &InventoryReference{SKU: "ABC-1", LocationID: "se"}, + CartID: "cart-1", + Quantity: 3, + TTL: 15 * time.Minute, + }) + if err != nil { + t.Fatalf("Reserve: %v", err) + } + avail, _ := p.Available(ctx, "ABC-1", "se") + if avail != 7 { + t.Fatalf("after reserve 3 of 10, expected 7 available, got %d", avail) + } + + // Release + if err := p.Release(ctx, "ABC-1", "se", "cart-1"); err != nil { + t.Fatalf("Release: %v", err) + } + avail, _ = p.Available(ctx, "ABC-1", "se") + if avail != 10 { + t.Fatalf("after release, expected 10 available, got %d", avail) + } +} + +func TestInMemoryInsufficientStock(t *testing.T) { + ctx := context.Background() + p := NewInMemoryReservationPolicy( + SKU("ABC-1"), LocationID("se"), int64(2), + ) + + err := p.Reserve(ctx, CartReserveRequest{ + InventoryReference: &InventoryReference{SKU: "ABC-1", LocationID: "se"}, + CartID: "cart-1", + Quantity: 5, + TTL: 15 * time.Minute, + }) + if err == nil { + t.Fatal("expected insufficient stock error, got nil") + } +} + +func TestInMemoryConcurrentReserves(t *testing.T) { + ctx := context.Background() + p := NewInMemoryReservationPolicy( + SKU("ABC-1"), LocationID("se"), int64(100), + ) + + errs := make(chan error, 10) + for i := 0; i < 10; i++ { + go func(n int) { + errs <- p.Reserve(ctx, CartReserveRequest{ + InventoryReference: &InventoryReference{SKU: "ABC-1", LocationID: "se"}, + CartID: CartID(string(rune('0' + n))), + Quantity: 10, + TTL: 15 * time.Minute, + }) + }(i) + } + + for i := 0; i < 10; i++ { + if err := <-errs; err != nil { + t.Errorf("concurrent reserve %d: %v", i, err) + } + } + + avail, _ := p.Available(ctx, "ABC-1", "se") + if avail != 0 { + t.Fatalf("after 10×10 reserves from 100, expected 0 available, got %d", avail) + } +} diff --git a/order/order.go b/order/order.go index 3f445fe..d687ce8 100644 --- a/order/order.go +++ b/order/order.go @@ -16,14 +16,18 @@ package order import "encoding/json" // Line is one ordered line item, reduced to what reactors need: the SKU, how -// many units sold, and the inventory location to commit against. Pricing/tax -// live in the order grain, not on this seam. +// many units sold, the inventory location to commit against, and whether the +// item is drop-ship (fulfilled by the supplier, never decremented from local +// stock). Pricing/tax live in the order grain, not on this seam. type Line struct { SKU string `json:"sku"` Quantity int `json:"quantity"` // Location is the inventory location / store id to decrement. Empty means // "use the order's Country" (central stock) — see Created.Country. Location string `json:"location,omitempty"` + // DropShip marks items fulfilled directly by the supplier. The inventory + // reactor must skip the stock decrement for these lines. + DropShip bool `json:"drop_ship,omitempty"` } // Created is the payload of an event.OrderCreated envelope. diff --git a/projection/doc.go b/projection/doc.go new file mode 100644 index 0000000..e6fa8db --- /dev/null +++ b/projection/doc.go @@ -0,0 +1,27 @@ +// Package projection is the domain-neutral substrate for the "snapshot stream + +// local index" seam: one authoritative producer publishes a framed +// begin → chunk* → end snapshot (plus per-batch deltas) on the bus, and N +// receivers each assemble their OWN cold-start-ready, mmap-backed .bin index +// from it. No shared volume; each consumer is independently eventually +// consistent. +// +// It was extracted from platform/catalog (which now re-exports these types for +// its Projection wire shape) so other domains can ride the same rails without +// depending on catalog. The pieces: +// +// - SnapshotWriter[T] / SnapshotReader[T] (snapshot.go) — the value-agnostic +// key→blob .bin format: sorted records, binary-search Get, full-walk Scan, +// mmap on Unix. Generic over the stored type T. +// - SnapshotHolder[T] (manager.go) — race-free atomic hot-reload of a reader +// under concurrent Get/Scan, with a drain-grace close-batching reaper. +// - Store[W, T] (store.go) — the consumer state machine: snapshot assembly, +// epoch gating, same-epoch delta overlay (with delete tombstones), and the +// completeness guard. Generic over the WIRE type W (decoded by a +// domain-supplied Decoder) and the STORED type T (projected by a Transform). +// - The bus framing contract (framing.go) — Meta* keys, snapshot phases, and +// the decoded Update[W] shape. +// +// The framing is fixed and shared; the on-the-wire payload schema is the +// domain's own concern, bridged by Decoder[W]. See +// docs/plan-projectionstore-generalization.md for why this boundary exists. +package projection diff --git a/projection/framing.go b/projection/framing.go new file mode 100644 index 0000000..c5d11f4 --- /dev/null +++ b/projection/framing.go @@ -0,0 +1,57 @@ +package projection + +// Bus framing contract (shared by producer + consumer). +// +// A projection stream carries two channels distinguished only by the event's +// Meta map. The producer sets these keys and the consumer reads them — one +// definition so the two never drift. +// +// - A FULL SNAPSHOT is a framed begin → chunk* → end stream, all tagged with one +// MetaEpoch. begin/end carry MetaCount (total entries) for completeness +// validation; chunks carry the entries. +// - A DELTA has NO MetaSnapshotPhase key, and MetaEpoch names the snapshot base +// it builds on (consumers ignore a delta older than the base they hold). +// +// The framing is domain-neutral. What rides inside each chunk/delta payload is +// a domain-specific wire schema, decoded by the Decoder the consumer supplies +// (see Store / Decoder). catalog re-exports these constants for wire-compat. +const ( + MetaSnapshotPhase = "snapshot" // begin|chunk|end; absent ⇒ delta + MetaEpoch = "epoch" // int64 (Unix-ms), monotonic across rebuilds + MetaSeq = "seq" // chunk sequence within a snapshot + MetaCount = "count" // total entries (begin/end only) + MetaTenant = "tenant" // namespace tag + + SnapshotBegin = "begin" + SnapshotChunk = "chunk" + SnapshotEnd = "end" +) + +// Update is the decoded form of one entry in a snapshot chunk or a delta. Key +// is the index key; Value is the domain wire value; Deleted marks a removal +// (the consumer tombstones the key in its overlay, or skips it in a snapshot). +// This is the in-memory shape the framing layer works with — the on-the-wire +// JSON schema is the domain's own concern, bridged by a Decoder. +type Update[W any] struct { + Key string + Value W + Deleted bool +} + +// Decoder turns one chunk/delta payload (the event's Payload bytes) into +// decoded Updates. Each domain defines its own wire schema and supplies the +// decoder; the Store owns only the framing/assembly around it. A decode error +// is surfaced from HandleFrame so the caller may NACK. +type Decoder[W any] func(payload []byte) ([]Update[W], error) + +// Transform projects a wire value W down to the consumer's stored type T. It is +// the one place a consumer declares which facts it actually needs (and any +// derivation), so its local .bin holds only that. Every value entering the +// store — snapshot, delta, and fallback — passes through it exactly once. +type Transform[W, T any] func(W) T + +// Fallback resolves a key the local store doesn't have yet (cold start before +// the first snapshot, or a key added since the last base). It returns a raw +// wire value W — the store applies the same Transform to it. Return (_, false) +// for a genuine miss. May be nil. +type Fallback[W any] func(key string) (W, bool) diff --git a/projection/helpers_test.go b/projection/helpers_test.go new file mode 100644 index 0000000..167c34c --- /dev/null +++ b/projection/helpers_test.go @@ -0,0 +1,54 @@ +// Local helpers for the snapshot / holder tests — production code does not see +// them. tProj is a stand-in stored value; the projection package is +// value-agnostic, so the tests use a small local struct instead of depending on +// catalog.Projection (the type these tests were extracted alongside). +package projection + +import "os" + +// tProj mirrors the fields the ported catalog snapshot tests exercise. Prices +// are plain int64 here (no money dependency in the generic package). +type tProj struct { + ID string `json:"id"` + SKU string `json:"sku"` + Title string `json:"title,omitempty"` + PriceIncVat int64 `json:"price_inc_vat"` + Currency string `json:"currency,omitempty"` + TaxClass string `json:"tax_class,omitempty"` + UpdatedAt int64 `json:"updated_at,omitempty"` +} + +func os2WriteFile(path string, data []byte, perm os.FileMode) error { + return os.WriteFile(path, data, perm) +} + +func os2Stat(path string) (os.FileInfo, error) { + return os.Stat(path) +} + +// appendP appends a tProj keyed by its own SKU — the producer/test pattern over +// the generic SnapshotWriter (Append takes an explicit key). +func appendP(w *SnapshotWriter[tProj], p tProj) error { + return w.Append(p.SKU, p) +} + +// makeProjection builds a tProj fixture. +func makeProjection(sku string, price int64) tProj { + return tProj{ + ID: "sku:" + sku, + SKU: sku, + Title: "Item " + sku, + PriceIncVat: price, + Currency: "SEK", + TaxClass: "standard", + UpdatedAt: 1, + } +} + +func repeat32(s string, n int) string { + out := make([]byte, 0, n*len(s)) + for i := 0; i < n; i++ { + out = append(out, s...) + } + return string(out) +} diff --git a/catalog/snapshot_manager.go b/projection/manager.go similarity index 89% rename from catalog/snapshot_manager.go rename to projection/manager.go index 50104a3..068262b 100644 --- a/catalog/snapshot_manager.go +++ b/projection/manager.go @@ -12,21 +12,20 @@ // The drain grace is retained as a close-BATCHING optimization: it defers and // coalesces reader closes onto the reaper tick instead of taking the exclusive // Lock on every swap. Pattern source: slask-finder's mmap+reload stores. -// See docs/inventory-shape.md for the prior art. // // Use: // -// holder := catalog.NewSnapshotHolder(30 * time.Second) // drain grace +// holder := projection.NewSnapshotHolder[T](30 * time.Second) // drain grace // if err := holder.Open("/var/lib/catalog/bygglagret.bin"); err != nil { ... } // // later, when a fresh snapshot lands: // if err := holder.Reload("/var/lib/catalog/bygglagret.bin"); err != nil { ... } -// // pulls the projection; always hits the current reader: -// p, ok := holder.Get(sku) +// // pulls the value; always hits the current reader: +// p, ok := holder.Get(key) // -// holder.Get / Len are safe for parallel use. Open and Reload are also +// holder.Get / Scan / Len are safe for parallel use. Open and Reload are also // safe for parallel use. Close stops the reaper goroutine and shuts the // current reader + drains down. -package catalog +package projection import ( "sync" @@ -48,10 +47,7 @@ type SnapshotHolder[T any] struct { // readMu makes reclamation race-free: Get holds RLock across the // Load-and-read of a reader, and a reader is only Close()d (munmap) under // Lock — so a reader can never be unmapped while a Get is reading it, even - // if that Get's goroutine was descheduled past the drain grace. The grace - // (below) is now only close-BATCHING — defer + coalesce closes off the swap - // path — NOT the safety mechanism it was before (which was best-effort and - // could SIGSEGV under a long enough scheduler pause). + // if that Get's goroutine was descheduled past the drain grace. readMu sync.RWMutex drainsMu sync.Mutex @@ -138,6 +134,20 @@ func (h *SnapshotHolder[T]) Get(sku string) (T, bool) { return r.Get(sku) } +// Scan walks every (key, value) of the current reader in key-sorted order. +// Return false from fn to stop early. The RLock is held for the whole scan so +// the reader can't be munmapped mid-walk — keep fn cheap (collect then sort +// outside the callback for ranked output). No-op when the holder is empty. +func (h *SnapshotHolder[T]) Scan(fn func(key string, v T) bool) { + h.readMu.RLock() + defer h.readMu.RUnlock() + r := h.current.Load() + if r == nil { + return + } + r.Scan(fn) +} + // Len returns the count of the current reader (0 if empty). func (h *SnapshotHolder[T]) Len() int { h.readMu.RLock() @@ -208,8 +218,6 @@ func (h *SnapshotHolder[T]) reapNow() { h.drainsMu.Unlock() for _, d := range reaped { // Lock briefly so the munmap can't race an in-flight Get on this reader. - // Gets that loaded this reader hold RLock and finish before the Lock is - // granted; Gets starting after see the swapped-in reader, never this one. h.readMu.Lock() _ = d.r.Close() h.readMu.Unlock() diff --git a/catalog/snapshot_manager_test.go b/projection/manager_test.go similarity index 78% rename from catalog/snapshot_manager_test.go rename to projection/manager_test.go index 4048848..dbe8b27 100644 --- a/catalog/snapshot_manager_test.go +++ b/projection/manager_test.go @@ -1,4 +1,4 @@ -package catalog +package projection import ( "fmt" @@ -7,8 +7,6 @@ import ( "sync/atomic" "testing" "time" - - "git.k6n.net/mats/platform/money" ) // TestSnapshotHolder_SwapRace storms the holder: 100 readers Get in tight loops @@ -17,13 +15,9 @@ import ( // // - No panic / SIGBUS — completion is the proof the drain grace never unmaps a // reader while an in-flight Get is reading it. -// - No torn reads — every hit decodes to a valid Projection (SKU matches, -// price > 0). A read of a munmapped page would crash or fail to decode. -// - No gap — since every snapshot version contains every SKU, a Get must never -// miss; misses would mean the holder briefly exposed an empty/nil reader -// mid-swap. (Exact "stale beyond drain grace" timing is intentionally not -// asserted — it's timing-flaky, and the no-torn-read + no-crash checks -// already cover the drain-safety property it stands in for.) +// - No torn reads — every hit decodes to a valid value (SKU matches, price > 0). +// - No gap — since every snapshot version contains every key, a Get must never +// miss; misses would mean the holder briefly exposed an empty/nil reader. func TestSnapshotHolder_SwapRace(t *testing.T) { dir := t.TempDir() @@ -31,11 +25,11 @@ func TestSnapshotHolder_SwapRace(t *testing.T) { const skuCount = 200 paths := make([]string, versions) for v := 0; v < versions; v++ { - w := NewSnapshotWriter[Projection]() + w := NewSnapshotWriter[tProj]() for i := 0; i < skuCount; i++ { sku := fmt.Sprintf("SKU-%03d", i) // price encodes the version (always > 0) so a torn read is detectable. - if err := w.Append(sku, Projection{ID: "sku:" + sku, SKU: sku, PriceIncVat: money.Cents(int64((v + 1) * 100))}); err != nil { + if err := w.Append(sku, tProj{ID: "sku:" + sku, SKU: sku, PriceIncVat: int64((v + 1) * 100)}); err != nil { t.Fatal(err) } } @@ -46,7 +40,7 @@ func TestSnapshotHolder_SwapRace(t *testing.T) { paths[v] = p } - h := NewSnapshotHolder[Projection](100 * time.Millisecond) + h := NewSnapshotHolder[tProj](100 * time.Millisecond) if err := h.Open(paths[0]); err != nil { t.Fatal(err) } @@ -111,7 +105,7 @@ func TestSnapshotHolder_SwapRace(t *testing.T) { t.Fatal("torn/garbage read or reload error during swap storm") } if m := atomic.LoadInt64(&misses); m != 0 { - t.Fatalf("got %d misses — holder exposed an empty reader mid-swap (every version has every SKU)", m) + t.Fatalf("got %d misses — holder exposed an empty reader mid-swap (every version has every key)", m) } if atomic.LoadInt64(&reads) == 0 { t.Fatal("no successful reads — test did not exercise the holder") @@ -125,12 +119,10 @@ type fakeCloser struct{ hits int } func (f *fakeCloser) Close() error { f.hits++; return nil } -// withFakeCloser swaps SnapshotReader's closer for our counter. Returns the -// reader and the counter; tests then call holder.Set(reloaded) and assert -// that close is invoked after the drain grace. -func withFakeCloser(t *testing.T, path string, fc *fakeCloser) *SnapshotReader[Projection] { +// withFakeCloser swaps SnapshotReader's closer for our counter. +func withFakeCloser(t *testing.T, path string, fc *fakeCloser) *SnapshotReader[tProj] { t.Helper() - r, err := ReadSnapshotFile[Projection](path) + r, err := ReadSnapshotFile[tProj](path) if err != nil { t.Fatal(err) } @@ -142,13 +134,11 @@ func withFakeCloser(t *testing.T, path string, fc *fakeCloser) *SnapshotReader[P // TestSnapshotHolderReloadDrainsOld — Set swaps the current reader; the // previous reader is queued. Before grace: still Get-able. After grace: // reaped and Close fires. -// -// We use a very short grace so the test doesn't sleep long. func TestSnapshotHolderReloadDrainsOld(t *testing.T) { dir := t.TempDir() - // 1) Build a valid first snapshot (skus A-1 / A-2). - w := NewSnapshotWriter[Projection]() + // 1) Build a valid first snapshot (keys A-1 / A-2). + w := NewSnapshotWriter[tProj]() if err := appendP(w, makeProjection("A-1", 100)); err != nil { t.Fatal(err) } @@ -162,8 +152,8 @@ func TestSnapshotHolderReloadDrainsOld(t *testing.T) { first := withFakeCloser(t, filepath.Join(dir, "first.bin"), firstCloser) defer first.Close() - // 2) Build a second snapshot (skus B-1). - w2 := NewSnapshotWriter[Projection]() + // 2) Build a second snapshot (key B-1). + w2 := NewSnapshotWriter[tProj]() if err := appendP(w2, makeProjection("B-1", 300)); err != nil { t.Fatal(err) } @@ -174,7 +164,7 @@ func TestSnapshotHolderReloadDrainsOld(t *testing.T) { second := withFakeCloser(t, filepath.Join(dir, "second.bin"), secondCloser) defer second.Close() - h := NewSnapshotHolder[Projection](50 * time.Millisecond) + h := NewSnapshotHolder[tProj](50 * time.Millisecond) defer h.Close() h.Set(first) if h.Len() != 2 { @@ -209,7 +199,7 @@ func TestSnapshotHolderReloadDrainsOld(t *testing.T) { // TestSnapshotHolderEmptyGet — holder with no current serves empty. func TestSnapshotHolderEmptyGet(t *testing.T) { - h := NewSnapshotHolder[Projection](100 * time.Millisecond) + h := NewSnapshotHolder[tProj](100 * time.Millisecond) defer h.Close() if _, ok := h.Get("anything"); ok { t.Error("empty holder should return false") @@ -220,12 +210,11 @@ func TestSnapshotHolderEmptyGet(t *testing.T) { } // TestSnapshotHolderCloseImmediatelyReaps — Close with pending drains -// should close them now (no grace wait). Confirms the drain queue is -// drained on Close. +// should close them now (no grace wait). func TestSnapshotHolderCloseImmediatelyReaps(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter[Projection]() + w := NewSnapshotWriter[tProj]() if err := appendP(w, makeProjection("X-1", 1)); err != nil { t.Fatal(err) } @@ -235,7 +224,7 @@ func TestSnapshotHolderCloseImmediatelyReaps(t *testing.T) { fc1 := &fakeCloser{} r1 := withFakeCloser(t, filepath.Join(dir, "x.bin"), fc1) - w2 := NewSnapshotWriter[Projection]() + w2 := NewSnapshotWriter[tProj]() if err := appendP(w2, makeProjection("Y-1", 2)); err != nil { t.Fatal(err) } @@ -245,7 +234,7 @@ func TestSnapshotHolderCloseImmediatelyReaps(t *testing.T) { fc2 := &fakeCloser{} r2 := withFakeCloser(t, filepath.Join(dir, "y.bin"), fc2) - h := NewSnapshotHolder[Projection](10 * time.Second) // long grace so reap wouldn't happen on its own + h := NewSnapshotHolder[tProj](10 * time.Second) // long grace so reap wouldn't happen on its own h.Set(r1) h.Set(r2) @@ -261,11 +250,10 @@ func TestSnapshotHolderCloseImmediatelyReaps(t *testing.T) { } // TestSnapshotHolderIdentityReload — Open the same path twice; current reader -// is replaced; DrainCount returns 1 (old queued). Old's data still served -// until grace elapses. +// is replaced; DrainCount returns 1 (old queued). func TestSnapshotHolderIdentityReload(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter[Projection]() + w := NewSnapshotWriter[tProj]() if err := appendP(w, makeProjection("ID-1", 1)); err != nil { t.Fatal(err) } @@ -273,7 +261,7 @@ func TestSnapshotHolderIdentityReload(t *testing.T) { t.Fatal(err) } - h := NewSnapshotHolder[Projection](10 * time.Second) + h := NewSnapshotHolder[tProj](10 * time.Second) defer h.Close() if err := h.Reload(filepath.Join(dir, "id.bin")); err != nil { t.Fatal(err) diff --git a/catalog/snapshot.go b/projection/snapshot.go similarity index 72% rename from catalog/snapshot.go rename to projection/snapshot.go index 285047c..f7daba4 100644 --- a/catalog/snapshot.go +++ b/projection/snapshot.go @@ -1,41 +1,43 @@ -// Snapshot reader/writer for the published catalog Projection. +// Snapshot reader/writer — a value-agnostic, mmap-friendly key→blob store. // -// Why a snapshot file at all? The bus is the live channel, but it doesn't -// carry the historical population on cold start. A snapshot file gives -// consumers (cart / checkout / order / UCP add-to-cart) a fast mmap-loaded -// cold-start substrate, so the first add-line request doesn't have to wait for -// the bus to replay. The bus then keeps it fresh. +// This is the domain-neutral substrate extracted from platform/catalog: the +// bus is the live channel, but it doesn't carry the historical population on +// cold start. A snapshot file gives a consumer a fast mmap-loaded cold-start +// substrate, so the first read doesn't have to wait for the bus to replay. The +// bus then keeps it fresh. catalog re-exports these types for its Projection; +// other domains (tracking popularity/suggestions) instantiate them over their +// own value type. // // Format v2 (little-endian, fixed-width, mmap-friendly): // // header [72 bytes] magic(8) version(4) reserved(4) count(8) // blobStart(8) recordStart(8) tenant(32, zero-padded) -// blobs variable count sequential JSON-encoded Projection bodies, no padding -// records count*44 each: sku(32, zero-padded) blobOffset(8) blobLength(4); -// sorted by SKU via the single compareSKU total order +// blobs variable count sequential JSON-encoded value bodies, no padding +// records count*44 each: key(32, zero-padded) blobOffset(8) blobLength(4); +// sorted by key via the single compareSKU total order // // Tenant field at offsets 40–71 (zero-padded) names the namespace this -// snapshot covers. Independent SKU schemas can produce Projections into -// different tenant files without conflict. Hot-reload across tenants stays -// safe under SnapshotHolder (snapshot_manager.go). +// snapshot covers. Independent key schemas can produce values into different +// tenant files without conflict. Hot-reload across tenants stays safe under +// SnapshotHolder (manager.go). // -// Sorting + binary search gives O(log N) per-SKU lookup with no on-disk hash -// table and no collisions. SKUs longer than 32 bytes are rejected at write -// time (Cart SKUs in this codebase are well under that). +// Sorting + binary search gives O(log N) per-key lookup with no on-disk hash +// table and no collisions. Keys longer than 32 bytes are rejected at write +// time. (Domains with longer natural keys — page paths, query strings — hash +// the key to a fixed-width digest and carry the original inside the value blob.) // // Memory model: the Reader holds one []byte for the entire file (mmap on -// Unix, ReadFile elsewhere). Per-SKU Get reads at most log2(N) record slots -// (each 44 B) and one JSON blob (~200 B). For 220 k projections the -// resident set is "pages touched this minute", not the whole file. +// Unix, ReadFile elsewhere). Per-key Get reads at most log2(N) record slots +// (each 44 B) and one JSON blob. Scan walks the sorted record section once. // // SAFETY: SnapshotReader alone is NOT safe across hot-reload. During reload, -// in-flight Get() calls on the OLD reader continue to work until disposal +// in-flight Get()/Scan() calls on the OLD reader continue to work until disposal // of the old reader is finalised; calling SnapshotReader.Close() concurrently // with Get() on the same reader risks unmapping pages-under-read (SIGBUS). -// Use SnapshotHolder (snapshot_manager.go) for any reload-bearing setup — -// it gives you an atomic.Pointer swap with a drain-grace (default 30s) so -// concurrent Gets on the OLD reader finish naturally before munmap fires. -package catalog +// Use SnapshotHolder (manager.go) for any reload-bearing setup — it gives you +// an atomic.Pointer swap with a drain-grace (default 30s) so concurrent Gets on +// the OLD reader finish naturally before munmap fires. +package projection import ( "encoding/binary" @@ -52,31 +54,28 @@ import ( const ( snapshotMagic uint64 = 0x3230564a53435050 // "PCSJ002" v2 (was 0x3130564a53435050 = v1) snapshotVersion uint32 = 2 - maxSKULen = 32 - maxTenantLen = 32 - headerSize = 8 + 4 + 4 + 8 + 8 + 8 + 32 // 72 bytes - recordSize = 32 + 8 + 4 // 44 bytes per record + maxSKULen = 32 + maxTenantLen = 32 + headerSize = 8 + 4 + 4 + 8 + 8 + 8 + 32 // 72 bytes + recordSize = 32 + 8 + 4 // 44 bytes per record ) var ( - ErrSnapshotMagic = errors.New("catalog: not a snapshot file (bad magic)") - ErrSnapshotVersion = errors.New("catalog: unsupported snapshot version") - ErrSnapshotTruncated = errors.New("catalog: snapshot truncated") - ErrSnapshotBadBlob = errors.New("catalog: snapshot record points past file") - ErrSnapshotOverflow = errors.New("catalog: snapshot record-section size overflows uint64") - ErrSKUTooLong = fmt.Errorf("catalog: SKU longer than %d bytes", maxSKULen) - ErrTenantTooLong = fmt.Errorf("catalog: tenant name longer than %d bytes", maxTenantLen) + ErrSnapshotMagic = errors.New("projection: not a snapshot file (bad magic)") + ErrSnapshotVersion = errors.New("projection: unsupported snapshot version") + ErrSnapshotTruncated = errors.New("projection: snapshot truncated") + ErrSnapshotBadBlob = errors.New("projection: snapshot record points past file") + ErrSnapshotOverflow = errors.New("projection: snapshot record-section size overflows uint64") + ErrSKUTooLong = fmt.Errorf("projection: key longer than %d bytes", maxSKULen) + ErrTenantTooLong = fmt.Errorf("projection: tenant name longer than %d bytes", maxTenantLen) ) // The snapshot is generic over the stored value T (see SnapshotWriter[T] / // SnapshotReader[T] below). The binary format, sorting, mmap, and bounds-check // logic is value-agnostic and lives in the unexported raw* layer; the generic -// wrappers only marshal T → blob (write) and blob → T (read). The producer -// stores catalog.Projection; each consumer stores its OWN projected type via a -// transform (see ProjectionStore[T] in consumer.go) — that's the per-consumer -// read-only projection from docs/catalog-projection.md. +// wrappers only marshal T → blob (write) and blob → T (read). -// rawSnapshotWriter is the value-agnostic writer: it indexes (sku → blob bytes). +// rawSnapshotWriter is the value-agnostic writer: it indexes (key → blob bytes). // Single-shot: append() all, then write() once (atomic rename). type rawSnapshotWriter struct { tenant string @@ -85,7 +84,7 @@ type rawSnapshotWriter struct { dirty bool } -// snapshotRecord is the on-disk layout: SKU zero-padded to 32 bytes, then +// snapshotRecord is the on-disk layout: key zero-padded to 32 bytes, then // an 8-byte offset (from file start) plus a 4-byte length for the JSON blob. type snapshotRecord struct { SKU [32]byte @@ -93,7 +92,7 @@ type snapshotRecord struct { BlobLength uint32 } -// append adds a pre-encoded blob under sku. ErrSKUTooLong if sku exceeds the +// append adds a pre-encoded blob under key. ErrSKUTooLong if key exceeds the // on-disk width. func (w *rawSnapshotWriter) append(sku string, blob []byte) error { if len(sku) > maxSKULen { @@ -112,8 +111,7 @@ func (w *rawSnapshotWriter) append(sku string, blob []byte) error { // SnapshotWriter[T] is single-shot: Append() all values, then Write() once // (atomic rename-into-place so readers never see a partial file). T is the -// stored value type (Projection for the producer; a consumer-specific type for -// a ProjectionStore[T]). +// stored value type. type SnapshotWriter[T any] struct { raw rawSnapshotWriter } @@ -132,16 +130,16 @@ func (w *SnapshotWriter[T]) WithTenant(tenant string) *SnapshotWriter[T] { return w } -// Append encodes v and indexes it under sku. Returns ErrSKUTooLong if sku +// Append encodes v and indexes it under key. Returns ErrSKUTooLong if key // exceeds the on-disk width; returns marshal errors as wrapped fmt.Errorf. -// SKU is explicit because T is opaque — the caller knows the key. +// key is explicit because T is opaque — the caller knows the key. func (w *SnapshotWriter[T]) Append(sku string, v T) error { if len(sku) > maxSKULen { return ErrSKUTooLong } jb, err := json.Marshal(v) if err != nil { - return fmt.Errorf("catalog: marshal snapshot value %q: %w", sku, err) + return fmt.Errorf("projection: marshal snapshot value %q: %w", sku, err) } return w.raw.append(sku, jb) } @@ -155,7 +153,7 @@ func (w *SnapshotWriter[T]) Write(dir, name string) (finalPath string, err error } // write serialises to a temp file in dir, fsync's, then renames into the -// final path. Records are sorted by SKU asc before serialisation so the +// final path. Records are sorted by key asc before serialisation so the // reader's binary search works. func (w *rawSnapshotWriter) write(dir, name string) (finalPath string, err error) { if w.dirty { @@ -233,12 +231,11 @@ func (w *rawSnapshotWriter) write(dir, name string) (finalPath string, err error return finalPath, nil } -// compareSKU defines the SINGLE TOTAL ORDER for SKUs across both Writer +// compareSKU defines the SINGLE TOTAL ORDER for keys across both Writer // (uses it for sort.Slice) and Reader (uses it for binary search). One // definition; invariant test verifies the on-disk result is sorted under -// it. Source SKUs SHOULD NOT contain embedded null bytes — the padded -// comparator treats them as terminators, which silently truncates an SKU. -// Realistically no SKU has NUL. +// it. Source keys SHOULD NOT contain embedded null bytes — the padded +// comparator treats them as terminators, which silently truncates a key. func compareSKU(padded, other []byte) int { // padded is a fixed-size zero-padded slot; trim trailing zeros. end := len(padded) @@ -272,16 +269,12 @@ func compareSKU(padded, other []byte) int { // compareRecordLookup is compareSKU specialised for Reader lookups so the // argument order matches the lex order: "padded" (fixed-slot) and "want" -// (query). -// -// compareSKU is symmetric so this is just an alias used by Reader code -// paths to make intent obvious. Any change to the total order here MUST be -// reflected in Writer.sort.Slice and verified by the invariant test. +// (query). compareSKU is symmetric so this is just an alias. func compareRecordLookup(padded, want []byte) int { return compareSKU(padded, want) } -// rawSnapshotReader is the value-agnostic reader: binary search by SKU, returns +// rawSnapshotReader is the value-agnostic reader: binary search by key, returns // the raw blob bytes. Concurrency-safe for parallel getRaw; NOT safe racing // Close with getRaw (see package doc — use SnapshotHolder for reload). type rawSnapshotReader struct { @@ -322,7 +315,7 @@ func NewSnapshotReader[T any](data []byte) (*SnapshotReader[T], error) { return &SnapshotReader[T]{raw: rr}, nil } -// Get returns the value for sku, or (zero, false) if absent / undecodable. +// Get returns the value for key, or (zero, false) if absent / undecodable. func (r *SnapshotReader[T]) Get(sku string) (T, bool) { var zero T blob, ok := r.raw.getRaw(sku) @@ -336,6 +329,22 @@ func (r *SnapshotReader[T]) Get(sku string) (T, bool) { return v, true } +// Scan walks every (key, value) in key-sorted order, calling fn for each. +// Return false from fn to stop early. A value that fails to decode is skipped +// (fn is not called for it). This is the enumeration / ranked-scan substrate: +// callers that need top-N-by-value collect into a slice and sort in memory — +// the on-disk order is by key, not value. NOT safe racing Close; drive it +// through SnapshotHolder.Scan under reload. +func (r *SnapshotReader[T]) Scan(fn func(key string, v T) bool) { + r.raw.scan(func(key string, blob []byte) bool { + var v T + if err := json.Unmarshal(blob, &v); err != nil { + return true // skip undecodable, keep going + } + return fn(key, v) + }) +} + // Len returns the number of values in this snapshot. func (r *SnapshotReader[T]) Len() int { return r.raw.Len() } @@ -379,7 +388,7 @@ func newRawSnapshotReader(data []byte) (*rawSnapshotReader, error) { blobStart := binary.LittleEndian.Uint64(data[24:32]) recordStart := binary.LittleEndian.Uint64(data[32:40]) if blobStart != uint64(headerSize) { - return nil, fmt.Errorf("catalog: blobStart=%d expected %d", blobStart, headerSize) + return nil, fmt.Errorf("projection: blobStart=%d expected %d", blobStart, headerSize) } // Multiplication overflow guard: for a malformed/tampered count, the @@ -402,10 +411,6 @@ func newRawSnapshotReader(data []byte) (*rawSnapshotReader, error) { // Per-record blob-offset bounds check: defends against malformed / // tampered files where the record section is intact but the blob // pointer lies past EOF. A linear pass; cheap (44 B × count at open). - // - // Avoid the naive blobOff+blobLen > dataLen check — blobOff can be - // MaxUint64 in tampered input and uint64 addition wraps, falsely - // passing the test. Split the check into overflow-safe parts. dataLen := uint64(len(data)) for i := uint64(0); i < count; i++ { recOff := recordStart + i*uint64(recordSize) @@ -439,9 +444,6 @@ func (r *rawSnapshotReader) Len() int { return int(r.count) } // Close releases the file handle and mmap region (if any). Idempotent — safe // to call from multiple goroutines (uses sync.Once under the hood). -// -// SAFETY: this is NOT safe against concurrent getRaw() — see package doc. Use -// SnapshotHolder for any reload-bearing workflow. func (r *rawSnapshotReader) Close() error { var err error r.closeOnce.Do(func() { @@ -452,7 +454,7 @@ func (r *rawSnapshotReader) Close() error { return err } -// getRaw returns the blob bytes for sku, or (nil, false) if not found. The +// getRaw returns the blob bytes for key, or (nil, false) if not found. The // returned slice aliases the mmap region — valid until Close. O(log N) binary // search on the sorted record section. func (r *rawSnapshotReader) getRaw(sku string) ([]byte, bool) { @@ -485,4 +487,21 @@ func (r *rawSnapshotReader) getRaw(sku string) ([]byte, bool) { return nil, false } - +// scan walks every record in key-sorted order, yielding (key, blob). Return +// false from fn to stop. The blob aliases the mmap region — valid until Close. +func (r *rawSnapshotReader) scan(fn func(key string, blob []byte) bool) { + for i := uint64(0); i < r.count; i++ { + recOff := int(r.recordStart) + int(i)*recordSize + recSKU := r.data[recOff : recOff+32] + end := len(recSKU) + for end > 0 && recSKU[end-1] == 0 { + end-- + } + key := string(recSKU[:end]) + blobLen := binary.LittleEndian.Uint32(r.data[recOff+40 : recOff+44]) + blobOff := binary.LittleEndian.Uint64(r.data[recOff+32 : recOff+40]) + if !fn(key, r.data[blobOff:blobOff+uint64(blobLen)]) { + return + } + } +} diff --git a/catalog/snapshot_other.go b/projection/snapshot_other.go similarity index 96% rename from catalog/snapshot_other.go rename to projection/snapshot_other.go index 996abf9..9183d30 100644 --- a/catalog/snapshot_other.go +++ b/projection/snapshot_other.go @@ -4,7 +4,7 @@ // identical to the unix mmap path — same reader code, same lookups — just // pays the full resident-set cost at Open. Production mmap-on-Linux/Darwin; // this exists for windows / plan9 / wasm builds. -package catalog +package projection import "os" diff --git a/catalog/snapshot_test.go b/projection/snapshot_test.go similarity index 76% rename from catalog/snapshot_test.go rename to projection/snapshot_test.go index 78ea39b..cccf41b 100644 --- a/catalog/snapshot_test.go +++ b/projection/snapshot_test.go @@ -1,4 +1,4 @@ -package catalog +package projection import ( "encoding/binary" @@ -9,18 +9,16 @@ import ( "path/filepath" "testing" "time" - - "git.k6n.net/mats/platform/money" ) -// TestSnapshotRoundTrip — write 1000 projections in shuffled order, read back -// and verify all 1000 are found at the right SKU. +// TestSnapshotRoundTrip — write 1000 values in shuffled order, read back +// and verify all 1000 are found at the right key. func TestSnapshotRoundTrip(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "snap.bin") const n = 1000 - w := NewSnapshotWriter[Projection]().WithTenant("default") + w := NewSnapshotWriter[tProj]().WithTenant("default") // intentionally shuffled input — sort is what reorders at Write. perm := rand.Perm(n) for _, i := range perm { @@ -40,7 +38,7 @@ func TestSnapshotRoundTrip(t *testing.T) { t.Errorf("Write returned %q expected %q", final, path) } - r, err := ReadSnapshotFile[Projection](path) + r, err := ReadSnapshotFile[tProj](path) if err != nil { t.Fatalf("ReadSnapshotFile: %v", err) } @@ -53,7 +51,7 @@ func TestSnapshotRoundTrip(t *testing.T) { t.Errorf("Tenant: got %q want %q", r.Tenant(), "default") } - // every SKU present and correct + // every key present and correct for _, i := range rand.Perm(n)[:50] { // probe 50 random lookups sku := fmt.Sprintf("SKU-%05d", i) got, ok := r.Get(sku) @@ -61,25 +59,72 @@ func TestSnapshotRoundTrip(t *testing.T) { t.Errorf("Get(%q) not found", sku) continue } - want := money.Cents(100 + (i%97)*100) + want := int64(100 + (i%97)*100) if got.PriceIncVat != want { t.Errorf("Get(%q) PriceIncVat: got %d want %d", sku, got.PriceIncVat, want) } } - // missing SKU + // missing key if _, ok := r.Get("NOPE-99999"); ok { - t.Error("Get on missing SKU should return ok=false") + t.Error("Get on missing key should return ok=false") + } +} + +// TestSnapshotScanWalksAllInKeyOrder — Scan visits every record exactly once in +// ascending key order, and early-return stops the walk. +func TestSnapshotScanWalksAllInKeyOrder(t *testing.T) { + dir := t.TempDir() + const n = 300 + w := NewSnapshotWriter[tProj]() + for _, i := range rand.Perm(n) { + sku := fmt.Sprintf("K-%04d", i) + if err := appendP(w, makeProjection(sku, int64(i))); err != nil { + t.Fatal(err) + } + } + if _, err := w.Write(dir, "scan.bin"); err != nil { + t.Fatal(err) + } + r, err := ReadSnapshotFile[tProj](filepath.Join(dir, "scan.bin")) + if err != nil { + t.Fatal(err) + } + defer r.Close() + + var keys []string + r.Scan(func(key string, v tProj) bool { + if v.SKU != key { + t.Errorf("Scan value/key mismatch: key=%q value.SKU=%q", key, v.SKU) + } + keys = append(keys, key) + return true + }) + if len(keys) != n { + t.Fatalf("Scan visited %d records, want %d", len(keys), n) + } + for i := 1; i < len(keys); i++ { + if keys[i-1] >= keys[i] { + t.Fatalf("Scan not ascending at %d: %q >= %q", i, keys[i-1], keys[i]) + } + } + + // early-return stops the walk + count := 0 + r.Scan(func(string, tProj) bool { + count++ + return count < 5 + }) + if count != 5 { + t.Errorf("Scan early-return visited %d, want 5", count) } } // TestSnapshotCompareSKUInvariant — verify the on-disk record section is // sorted under the same compareSKU total order that BinarySearch uses. -// If sort.Slice and Reader.Get ever drift apart, lookups silently miss -// — this test pins that invariant. func TestSnapshotCompareSKUInvariant(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter[Projection]() + w := NewSnapshotWriter[tProj]() // deliberately inverted order after Append (so sort has work to do) perm := rand.Perm(500) for _, i := range perm { @@ -91,13 +136,13 @@ func TestSnapshotCompareSKUInvariant(t *testing.T) { if _, err := w.Write(dir, "snap.bin"); err != nil { t.Fatal(err) } - r, err := ReadSnapshotFile[Projection](filepath.Join(dir, "snap.bin")) + r, err := ReadSnapshotFile[tProj](filepath.Join(dir, "snap.bin")) if err != nil { t.Fatal(err) } defer r.Close() - // Walk every record slot, harvest SKU bytes from the pad-then-trim + // Walk every record slot, harvest key bytes from the pad-then-trim // form, and assert non-decreasing order under compareSKU. prevSKU := "" for i := 0; i < r.Len(); i++ { @@ -127,18 +172,17 @@ func runesToPadded32(s string) []byte { } // TestSnapshotTenantRoundTrip — Writer.WithTenant lands in the header and -// Reader surfaces it. Without this, one file per tenant is just a convention -// with no on-disk evidence. +// Reader surfaces it. func TestSnapshotTenantRoundTrip(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter[Projection]().WithTenant("bygglagret") + w := NewSnapshotWriter[tProj]().WithTenant("bygglagret") if err := appendP(w, makeProjection("ABC-1", 100)); err != nil { t.Fatal(err) } if _, err := w.Write(dir, "snap.bin"); err != nil { t.Fatal(err) } - r, err := ReadSnapshotFile[Projection](filepath.Join(dir, "snap.bin")) + r, err := ReadSnapshotFile[tProj](filepath.Join(dir, "snap.bin")) if err != nil { t.Fatal(err) } @@ -148,14 +192,14 @@ func TestSnapshotTenantRoundTrip(t *testing.T) { } // Empty tenant writes zero-padded. - w2 := NewSnapshotWriter[Projection]() + w2 := NewSnapshotWriter[tProj]() if err := appendP(w2, makeProjection("ABC-2", 200)); err != nil { t.Fatal(err) } if _, err := w2.Write(dir, "snap2.bin"); err != nil { t.Fatal(err) } - r2, err := ReadSnapshotFile[Projection](filepath.Join(dir, "snap2.bin")) + r2, err := ReadSnapshotFile[tProj](filepath.Join(dir, "snap2.bin")) if err != nil { t.Fatal(err) } @@ -168,7 +212,7 @@ func TestSnapshotTenantRoundTrip(t *testing.T) { // TestSnapshotTenantTooLong — over-long tenant name rejected at Write. func TestSnapshotTenantTooLong(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter[Projection]().WithTenant(string(make([]byte, 33))) + w := NewSnapshotWriter[tProj]().WithTenant(string(make([]byte, 33))) if err := appendP(w, makeProjection("ABC", 1)); err != nil { t.Fatal(err) } @@ -181,7 +225,7 @@ func TestSnapshotTenantTooLong(t *testing.T) { // TestSnapshotCountOverflowRejected — tampered count that would cause // recordStart + records*recordSize to overflow uint64 must be rejected -// with ErrSnapshotOverflow (formerly a silent wraparound vulnerability). +// with ErrSnapshotOverflow. func TestSnapshotCountOverflowRejected(t *testing.T) { dir := t.TempDir() path := filepath.Join(dir, "over.bin") @@ -196,7 +240,7 @@ func TestSnapshotCountOverflowRejected(t *testing.T) { if err := os2WriteFile(path, bs, 0o644); err != nil { t.Fatal(err) } - if _, err := ReadSnapshotFile[Projection](path); !errors.Is(err, ErrSnapshotOverflow) { + if _, err := ReadSnapshotFile[tProj](path); !errors.Is(err, ErrSnapshotOverflow) { t.Errorf("overflow count rejected with %v, want ErrSnapshotOverflow", err) } } @@ -205,8 +249,8 @@ func TestSnapshotCountOverflowRejected(t *testing.T) { // leaves a half-written snap.bin visible to readers. func TestSnapshotAtomicSwap(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter[Projection]() - mustAppend := func(w *SnapshotWriter[Projection], sku string, price int64) { + w := NewSnapshotWriter[tProj]() + mustAppend := func(w *SnapshotWriter[tProj], sku string, price int64) { t.Helper() if err := appendP(w, makeProjection(sku, price)); err != nil { t.Fatalf("Append %s: %v", sku, err) @@ -223,13 +267,13 @@ func TestSnapshotAtomicSwap(t *testing.T) { } // replace - w2 := NewSnapshotWriter[Projection]() + w2 := NewSnapshotWriter[tProj]() mustAppend(w2, "ZZZ", 200) if _, err := w2.Write(dir, "snap.bin"); err != nil { t.Fatal(err) } - r, err := ReadSnapshotFile[Projection](final) + r, err := ReadSnapshotFile[tProj](final) if err != nil { t.Fatal(err) } @@ -251,7 +295,7 @@ func TestSnapshotRejectBadInput(t *testing.T) { if err := os2WriteFile(path, []byte{}, 0o644); err != nil { t.Fatal(err) } - if _, err := ReadSnapshotFile[Projection](path); !errors.Is(err, ErrSnapshotTruncated) { + if _, err := ReadSnapshotFile[tProj](path); !errors.Is(err, ErrSnapshotTruncated) { t.Errorf("empty file rejected with %v, want ErrSnapshotTruncated", err) } @@ -265,7 +309,7 @@ func TestSnapshotRejectBadInput(t *testing.T) { if err := os2WriteFile(path, wrongMagic, 0o644); err != nil { t.Fatal(err) } - if _, err := ReadSnapshotFile[Projection](path); !errors.Is(err, ErrSnapshotMagic) { + if _, err := ReadSnapshotFile[tProj](path); !errors.Is(err, ErrSnapshotMagic) { t.Errorf("wrong magic rejected with %v, want ErrSnapshotMagic", err) } @@ -274,7 +318,7 @@ func TestSnapshotRejectBadInput(t *testing.T) { if err := os2WriteFile(path, buildValidHeader(5), 0o644); err != nil { t.Fatal(err) } - if _, err := ReadSnapshotFile[Projection](path); !errors.Is(err, ErrSnapshotTruncated) { + if _, err := ReadSnapshotFile[tProj](path); !errors.Is(err, ErrSnapshotTruncated) { t.Errorf("truncated snapshot rejected with %v, want ErrSnapshotTruncated", err) } @@ -284,7 +328,7 @@ func TestSnapshotRejectBadInput(t *testing.T) { if err := os2WriteFile(path, tampered, 0o644); err != nil { t.Fatal(err) } - if _, err := ReadSnapshotFile[Projection](path); !errors.Is(err, ErrSnapshotBadBlob) { + if _, err := ReadSnapshotFile[tProj](path); !errors.Is(err, ErrSnapshotBadBlob) { t.Errorf("tampered-blob rejected with %v, want ErrSnapshotBadBlob", err) } } @@ -292,11 +336,11 @@ func TestSnapshotRejectBadInput(t *testing.T) { // TestSnapshotEmpty — empty (count=0) snapshot is valid; Get returns false. func TestSnapshotEmpty(t *testing.T) { dir := t.TempDir() - w := NewSnapshotWriter[Projection]() + w := NewSnapshotWriter[tProj]() if _, err := w.Write(dir, "empty.bin"); err != nil { t.Fatal(err) } - r, err := ReadSnapshotFile[Projection](filepath.Join(dir, "empty.bin")) + r, err := ReadSnapshotFile[tProj](filepath.Join(dir, "empty.bin")) if err != nil { t.Fatal(err) } @@ -309,49 +353,48 @@ func TestSnapshotEmpty(t *testing.T) { } } -// TestSnapshotSKU32 — SKU exactly 32 chars (the boundary case) must succeed. +// TestSnapshotSKU32 — key exactly 32 chars (the boundary case) must succeed. func TestSnapshotSKU32(t *testing.T) { dir := t.TempDir() sku := "SKU-" + repeat32("X", 32-4) // 4 + 28 = 32 chars if len(sku) != 32 { t.Fatalf("test bug: sku length %d", len(sku)) } - w := NewSnapshotWriter[Projection]() + w := NewSnapshotWriter[tProj]() if err := appendP(w, makeProjection(sku, 1)); err != nil { - t.Fatalf("Append 32-char SKU: %v", err) + t.Fatalf("Append 32-char key: %v", err) } if _, err := w.Write(dir, "snap.bin"); err != nil { t.Fatal(err) } - r, err := ReadSnapshotFile[Projection](filepath.Join(dir, "snap.bin")) + r, err := ReadSnapshotFile[tProj](filepath.Join(dir, "snap.bin")) if err != nil { t.Fatal(err) } defer r.Close() if _, ok := r.Get(sku); !ok { - t.Error("32-char SKU Get returned false") + t.Error("32-char key Get returned false") } } // TestSnapshotSKU33Rejected — one char over should error at Append. func TestSnapshotSKU33Rejected(t *testing.T) { - w := NewSnapshotWriter[Projection]() + w := NewSnapshotWriter[tProj]() sku := repeat32("X", 33) err := appendP(w, makeProjection(sku, 1)) if !errors.Is(err, ErrSKUTooLong) { - t.Errorf("33-char SKU Append: got %v want ErrSKUTooLong", err) + t.Errorf("33-char key Append: got %v want ErrSKUTooLong", err) } } -// TestSnapshotGetSpeed — probe thousands of random SKU lookups to give a -// sense of the binary-search cost (not a benchmark per se). +// TestSnapshotGetSpeed — probe thousands of random key lookups. func TestSnapshotGetSpeed(t *testing.T) { if testing.Short() { t.Skip("short mode") } dir := t.TempDir() const n = 5000 - w := NewSnapshotWriter[Projection]() + w := NewSnapshotWriter[tProj]() skus := make([]string, n) for i := 0; i < n; i++ { skus[i] = fmt.Sprintf("Z%06d", i) @@ -362,7 +405,7 @@ func TestSnapshotGetSpeed(t *testing.T) { if _, err := w.Write(dir, "snap.bin"); err != nil { t.Fatal(err) } - r, err := ReadSnapshotFile[Projection](filepath.Join(dir, "snap.bin")) + r, err := ReadSnapshotFile[tProj](filepath.Join(dir, "snap.bin")) if err != nil { t.Fatal(err) } @@ -429,20 +472,7 @@ func TestSnapshotRejectBadVersion(t *testing.T) { if err := os2WriteFile(path, bs, 0o644); err != nil { t.Fatal(err) } - if _, err := ReadSnapshotFile[Projection](path); !errors.Is(err, ErrSnapshotVersion) { + if _, err := ReadSnapshotFile[tProj](path); !errors.Is(err, ErrSnapshotVersion) { t.Errorf("v2-not-supported header rejected with %v, want ErrSnapshotVersion", err) } } - -// makeProjection helper repeated here for the new tests. -func makeProjection(sku string, price int64) Projection { - return Projection{ - ID: "sku:" + sku, - SKU: sku, - Title: "Item " + sku, - PriceIncVat: money.Cents(price), - Currency: "SEK", - TaxClass: "standard", - UpdatedAt: time.Now().UnixMilli(), - } -} diff --git a/catalog/snapshot_unix.go b/projection/snapshot_unix.go similarity index 85% rename from catalog/snapshot_unix.go rename to projection/snapshot_unix.go index 7fb0e04..a5b4df1 100644 --- a/catalog/snapshot_unix.go +++ b/projection/snapshot_unix.go @@ -3,9 +3,9 @@ // mmap-on-unix loader for SnapshotReader. The file handle stays alive for // the lifetime of the mmap region (the kernel needs it pinned). Closer // munmaps AND closes the fd. Uses MADV_RANDOM because the dominant access -// pattern is binary-search-by-SKU — touching random record slots, not +// pattern is binary-search-by-key — touching random record slots, not // streaming the file. -package catalog +package projection import ( "fmt" @@ -39,10 +39,10 @@ func loadSnapshotBytes(path string) (data []byte, closer func() error, err error data, mmapErr := unix.Mmap(int(f.Fd()), 0, int(size), unix.PROT_READ, unix.MAP_SHARED) if mmapErr != nil { _ = f.Close() - return nil, nil, fmt.Errorf("catalog: mmap %s: %w", path, mmapErr) + return nil, nil, fmt.Errorf("projection: mmap %s: %w", path, mmapErr) } - // Binary-search by SKU = random page access; MADV_RANDOM tells the + // Binary-search by key = random page access; MADV_RANDOM tells the // kernel NOT to do read-ahead — saves I/O on cold layers. if maErr := unix.Madvise(data, unix.MADV_RANDOM); maErr != nil { // non-fatal; degrade gracefully diff --git a/projection/store.go b/projection/store.go new file mode 100644 index 0000000..45bcfbe --- /dev/null +++ b/projection/store.go @@ -0,0 +1,346 @@ +package projection + +// Store — the consumer half of the projection seam. A service feeds it the +// snapshot/delta frames it receives off the bus and reads facts back by key. +// The store owns a LOCAL read-only projection: +// +// - A full snapshot stream (begin → chunk* → end, one MetaEpoch) is assembled +// into the store's OWN .bin (SnapshotWriter), atomically swapped in as the +// mmap base (SnapshotHolder). The base is the cold-start substrate — it +// survives a restart (reopened from disk), so the consumer serves reads +// immediately without waiting for a bus replay. +// - Per-batch deltas of the SAME epoch are applied to an in-memory overlay on +// top of the base (overlay wins; a delete leaves a tombstone that shadows the +// base and the fallback). The next snapshot folds the overlay into a new base. +// +// Read order on Get: overlay → mmap base → optional caller fallback. Scan walks +// overlay + base (overlay wins, tombstones hidden) for enumeration / ranked +// output. +// +// The store is generic over the WIRE value W (what rides on the bus, decoded by +// the supplied Decoder) and the STORED value T (what the Transform projects W +// down to, held in the .bin). For an identity consumer W == T with an identity +// transform; catalog uses W = catalog.Projection. +// +// No shared volume: each consumer builds its own base from the bus. Concurrency: +// HandleFrame is single-writer (drive it from one delivery goroutine); Get/Scan +// are safe from many readers. + +import ( + "fmt" + "log" + "os" + "path/filepath" + "strconv" + "strings" + "sync" +) + +// Store is a consumer's local, cold-start-ready projection cache of T, fed from +// a bus stream of wire values W. +type Store[W, T any] struct { + dir, name string + decode Decoder[W] + transform Transform[W, T] + fallback Fallback[W] + holder *SnapshotHolder[T] // mmap base; atomic-swap + drain-grace + + mu sync.RWMutex + epoch int64 // epoch of the live base + overlay map[string]overlayEntry[T] // same-epoch deltas layered on the base + + // in-progress snapshot assembly (one epoch at a time) + building *SnapshotWriter[T] + buildingEpoch int64 + buildingSeen int +} + +type overlayEntry[T any] struct { + val T + deleted bool +} + +// NewStore opens a store backed by dir/name (its own .bin). decode and +// transform are REQUIRED: decode parses the domain wire schema from a frame +// payload; transform projects each wire value to the stored shape. If a base +// from a previous run is present it is reopened immediately (cold-start ready); +// its epoch is read from the sidecar so same-epoch deltas can resume. fallback +// may be nil. Close() when done. +func NewStore[W, T any](dir, name string, decode Decoder[W], transform Transform[W, T], fallback Fallback[W]) (*Store[W, T], error) { + if dir == "" || name == "" { + return nil, fmt.Errorf("projection: Store needs dir and name") + } + if decode == nil { + return nil, fmt.Errorf("projection: Store needs a decoder (declare the domain wire schema)") + } + if transform == nil { + return nil, fmt.Errorf("projection: Store needs a transform (declare which facts this consumer stores)") + } + s := &Store[W, T]{ + dir: dir, + name: name, + decode: decode, + transform: transform, + fallback: fallback, + holder: NewSnapshotHolder[T](0), + overlay: make(map[string]overlayEntry[T]), + } + path := filepath.Join(dir, name) + if err := s.holder.Open(path); err == nil { + s.epoch = readEpochSidecar(path) + log.Printf("projection: Store reopened base %s (epoch=%d, %d entries)", path, s.epoch, s.holder.Len()) + } else if !os.IsNotExist(err) { + // A corrupt/partial base on disk shouldn't crash boot — start empty and + // wait for the next snapshot; reads fall back meanwhile. + log.Printf("projection: Store ignoring unreadable base %s: %v", path, err) + } + return s, nil +} + +// HandleFrame applies one projection event (snapshot frame or delta), routed by +// its Meta. Drive this from a single delivery goroutine. A returned error means +// the payload was undecodable (the caller may NACK); a frame that simply +// doesn't apply (stale epoch, wrong assembly state) returns nil so the delivery +// is acked. +func (s *Store[W, T]) HandleFrame(meta map[string]string, payload []byte) error { + epoch, _ := strconv.ParseInt(meta[MetaEpoch], 10, 64) + switch meta[MetaSnapshotPhase] { + case SnapshotBegin: + return s.beginSnapshot(epoch, meta[MetaTenant]) + case SnapshotChunk: + return s.chunkSnapshot(epoch, payload) + case SnapshotEnd: + return s.endSnapshot(epoch, meta) + case "": + return s.applyDelta(epoch, payload) + default: + return nil // unknown phase — ignore forward-compatibly + } +} + +func (s *Store[W, T]) beginSnapshot(epoch int64, tenant string) error { + s.mu.Lock() + defer s.mu.Unlock() + if epoch <= s.epoch { + return nil // we already hold this base or newer + } + // Start (or restart) assembly for this epoch. A new begin supersedes any + // half-built snapshot (e.g. a producer that aborted mid-stream). + s.building = NewSnapshotWriter[T]().WithTenant(tenant) + s.buildingEpoch = epoch + s.buildingSeen = 0 + return nil +} + +func (s *Store[W, T]) chunkSnapshot(epoch int64, payload []byte) error { + ups, err := s.decode(payload) + if err != nil { + return fmt.Errorf("projection: decode snapshot chunk: %w", err) + } + s.mu.Lock() + defer s.mu.Unlock() + if s.building == nil || epoch != s.buildingEpoch { + return nil // not assembling this epoch + } + for _, u := range ups { + if u.Deleted || u.Key == "" { + continue // a snapshot carries live entries only + } + // Transform applied here — the base stores the consumer's shape. + if err := s.building.Append(u.Key, s.transform(u.Value)); err != nil { + log.Printf("projection: Store skip %q in snapshot: %v", u.Key, err) + continue + } + s.buildingSeen++ + } + return nil +} + +func (s *Store[W, T]) endSnapshot(epoch int64, meta map[string]string) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.building == nil || epoch != s.buildingEpoch { + return nil + } + w := s.building + s.building = nil + + // Completeness guard: if begin/end advertised a count and we appended fewer + // (a lost chunk), DON'T swap — keep the old base and wait for the next + // snapshot. Serving an incomplete projection silently is worse than stale. + if cs := meta[MetaCount]; cs != "" { + if want, perr := strconv.Atoi(cs); perr == nil && want != s.buildingSeen { + log.Printf("projection: Store snapshot epoch=%d incomplete (%d/%d) — discarding, keeping base epoch=%d", + epoch, s.buildingSeen, want, s.epoch) + return nil + } + } + + path, err := w.Write(s.dir, s.name) + if err != nil { + return fmt.Errorf("projection: write snapshot base: %w", err) + } + // Persist the epoch sidecar (atomic) BEFORE swapping the in-memory reader, so + // a reopened store's epoch always matches the .bin already on disk (w.Write + // renamed the new file into place above). Writing it after Reload would leave + // a crash window where the new .bin carries a stale/absent epoch. + writeEpochSidecar(path, epoch) + if err := s.holder.Reload(path); err != nil { + return fmt.Errorf("projection: reload snapshot base: %w", err) + } + s.epoch = epoch + s.overlay = make(map[string]overlayEntry[T]) // base now includes everything ≤ epoch + log.Printf("projection: Store snapshot applied — base swapped epoch=%d (%d entries)", epoch, s.buildingSeen) + return nil +} + +func (s *Store[W, T]) applyDelta(epoch int64, payload []byte) error { + ups, err := s.decode(payload) + if err != nil { + return fmt.Errorf("projection: decode delta: %w", err) + } + s.mu.Lock() + defer s.mu.Unlock() + switch { + case s.building != nil: + // Under FIFO delivery a delta won't interleave a snapshot stream; if it + // does, the snapshot is authoritative for its epoch, so drop the delta. + return nil + case epoch < s.epoch: + return nil // stale — predates our base + case epoch > s.epoch: + // CONTRACT: a future-epoch delta is DROPPED, not buffered. Under FIFO + // per-queue delivery begin(N) always precedes delta(N), so this only + // happens after a reconnect that missed the whole snapshot — which a + // buffer couldn't reassemble anyway. Recovery is the next snapshot of + // this epoch; the consumer's fallback covers reads until then. + return nil + } + var upserts, deletes int + for _, u := range ups { + if u.Key == "" { + continue + } + if u.Deleted { + s.overlay[u.Key] = overlayEntry[T]{deleted: true} + deletes++ + continue + } + // Transform applied here too — overlay holds the consumer's shape. + s.overlay[u.Key] = overlayEntry[T]{val: s.transform(u.Value)} + upserts++ + } + if upserts+deletes > 0 { + log.Printf("projection: Store delta applied epoch=%d (+%d/-%d, overlay=%d)", + epoch, upserts, deletes, len(s.overlay)) + } + return nil +} + +// Get resolves a value by key: overlay (incl. tombstone) → mmap base → +// fallback. A tombstoned key returns (zero, false) and never reaches the +// fallback. A fallback hit is run through the same Transform. +func (s *Store[W, T]) Get(sku string) (T, bool) { + var zero T + s.mu.RLock() + e, ok := s.overlay[sku] + s.mu.RUnlock() + if ok { + if e.deleted { + return zero, false + } + return e.val, true + } + if v, ok := s.holder.Get(sku); ok { + return v, true + } + if s.fallback != nil { + if w, ok := s.fallback(sku); ok { + return s.transform(w), true + } + } + return zero, false +} + +// Scan walks every live (key, value) — overlay first (overlay wins; tombstones +// hidden), then the mmap base minus any key the overlay already covered. Return +// false from fn to stop early. The fallback is NOT consulted (it resolves +// single misses, not enumeration). Callers that need top-N-by-value collect and +// sort in memory — on-disk order is by key. Held under RLock for the whole +// walk, so it blocks a concurrent snapshot swap; keep fn cheap. +func (s *Store[W, T]) Scan(fn func(key string, v T) bool) { + s.mu.RLock() + defer s.mu.RUnlock() + seen := make(map[string]struct{}, len(s.overlay)) + for k, e := range s.overlay { + seen[k] = struct{}{} + if e.deleted { + continue + } + if !fn(k, e.val) { + return + } + } + s.holder.Scan(func(k string, v T) bool { + if _, ok := seen[k]; ok { + return true // overlay already emitted (or tombstoned) this key + } + return fn(k, v) + }) +} + +// IsDeleted reports whether the overlay holds a tombstone for key — a delete +// seen on the bus since the current base. +func (s *Store[W, T]) IsDeleted(sku string) bool { + s.mu.RLock() + defer s.mu.RUnlock() + e, ok := s.overlay[sku] + return ok && e.deleted +} + +// Epoch returns the epoch of the current base (0 if none yet). +func (s *Store[W, T]) Epoch() int64 { + s.mu.RLock() + defer s.mu.RUnlock() + return s.epoch +} + +// Stats returns observability counters (base size, overlay size, epoch). +func (s *Store[W, T]) Stats() (epoch int64, baseCount, overlayCount int) { + s.mu.RLock() + defer s.mu.RUnlock() + return s.epoch, s.holder.Len(), len(s.overlay) +} + +// Close releases the mmap base. +func (s *Store[W, T]) Close() error { return s.holder.Close() } + +// ── epoch sidecar ──────────────────────────────────────────────────────────── +// The .bin header has no room for the epoch, so it's persisted next to it as +// ".epoch" — read on boot so a reopened base can resume same-epoch deltas. + +func epochSidecarPath(binPath string) string { return binPath + ".epoch" } + +// writeEpochSidecar writes the epoch atomically (temp+rename) so a crash mid-write +// can't leave a torn file that parses as a bogus epoch. +func writeEpochSidecar(binPath string, epoch int64) { + p := epochSidecarPath(binPath) + tmp := p + ".tmp" + if err := os.WriteFile(tmp, []byte(strconv.FormatInt(epoch, 10)), 0o644); err != nil { + log.Printf("projection: Store epoch sidecar write: %v", err) + return + } + if err := os.Rename(tmp, p); err != nil { + log.Printf("projection: Store epoch sidecar rename: %v", err) + _ = os.Remove(tmp) + } +} + +func readEpochSidecar(binPath string) int64 { + b, err := os.ReadFile(epochSidecarPath(binPath)) + if err != nil { + return 0 + } + n, _ := strconv.ParseInt(strings.TrimSpace(string(b)), 10, 64) + return n +} diff --git a/shipping/shipping.go b/shipping/shipping.go new file mode 100644 index 0000000..2c4b08e --- /dev/null +++ b/shipping/shipping.go @@ -0,0 +1,140 @@ +// Package shipping is the platform's shared shipping seam: the cross-module +// vocabulary for delivery options and the Provider interface. It holds only the +// wire value types plus the interface, so the producer (a shipping adapter like +// PostNord) and every consumer (checkout, cart, backoffice) agree on the types. +// It imports only the stdlib — never an app module, never a transport library. +// +// Dependency rule: packages under platform/ import only third-party libraries +// and other platform/ packages — never an application module. Application +// modules depend downward onto platform. This is what keeps it a stable seam. +// +// See platform/inventory for the sibling pattern (value types + interface in +// one stdlib-only package, test adapter in _test.go). +package shipping + +import ( + "context" + "encoding/json" + "errors" +) + +// ── Provider interface ───────────────────────────────────────────────────── + +// Provider is the cross-module seam for shipping delivery options. Checkout +// calls GetShippingOptions to present available delivery methods to the +// customer. Every implementation behind this interface is free to choose its +// internal provider (PostNord, DHL, Bring, etc.) — the consumer never needs +// to know. +// +// This is the policy-first seam: shipping selection policy lives here; the +// provider-specific HTTP/REST details live behind it. +type Provider interface { + // GetShippingOptions returns available delivery options for a recipient + // address. Returns an empty slice (not an error) when the provider has no + // matching options. Returns ErrNoOptions when no provider in a multi-provider + // setup could serve the address. + GetShippingOptions(ctx context.Context, req GetOptionsRequest) ([]DeliveryOptionGroup, error) +} + +// ── Request types ───────────────────────────────────────────────────────── + +// GetOptionsRequest is what a caller provides to get shipping options. +type GetOptionsRequest struct { + // Country is the ISO 3166-1 alpha-2 recipient country (e.g. "SE", "NO"). + Country string `json:"country"` + // PostCode is the recipient postal code. + PostCode string `json:"post_code"` + // Locale is the IETF BCP 47 language tag for descriptions (e.g. "sv", "en"). + // Empty means the provider chooses a default. + Locale string `json:"locale,omitempty"` + // Items optionally describes the cart items for dimension/weight-based + // filtering. May be nil or empty for providers that don't need item data. + Items []OptionItem `json:"items,omitempty"` +} + +// OptionItem describes one line item for shipping calculations (weight, volume, +// category-based restrictions). +type OptionItem struct { + SKU string `json:"sku"` + Quantity int `json:"quantity"` + // WeightGrams is the item weight in grams, if known. + WeightGrams int `json:"weight_grams,omitempty"` +} + +// ── Response types ──────────────────────────────────────────────────────── + +// DeliveryOptionGroup groups shipping options by fulfilment method (e.g. home +// delivery, pickup point, parcel locker). A provider returns one or more +// groups; each group has one default option and optional alternatives (e.g. +// different pickup point locations). +type DeliveryOptionGroup struct { + // Type is the fulfilment-method category: "home", "pickup", "locker", + // "mailbox", etc. + Type string `json:"type"` + // Name is the display name for the group (e.g. "Home delivery", + // "Pickup point"). Provider-localised to the request Locale. + Name string `json:"name"` + // DefaultOption is the recommended option within this group. + DefaultOption DeliveryOption `json:"default_option"` + // AdditionalOptions are alternatives within the same group (e.g. other + // pickup point locations, different time slots). + AdditionalOptions []DeliveryOption `json:"additional_options,omitempty"` +} + +// DeliveryOption is one selectable shipping method or service point. +type DeliveryOption struct { + // ID is the provider-scoped unique identifier for this option. Used when + // the caller confirms the selection (e.g. SetCartDelivery). + ID string `json:"id"` + // Name is a short display name (e.g. "Home delivery", "PostNord Storgatan"). + Name string `json:"name"` + // Description is a longer description for the checkout UI. + Description string `json:"description,omitempty"` + // DeliveryDays is a human-readable estimate (e.g. "2-3 working days", + // "1-2 days"). Provider-localised. + DeliveryDays string `json:"delivery_days,omitempty"` + // PickupLocation is set when this option is a physical pickup point. + PickupLocation *PickupLocation `json:"pickup_location,omitempty"` +} + +// PickupLocation describes a physical pickup point (store, locker, post office). +type PickupLocation struct { + ID string `json:"id"` + Name string `json:"name"` + Address Address `json:"address"` +} + +// Address is a simplified postal address for delivery or pickup. +type Address struct { + Street string `json:"street"` + Street2 string `json:"street2,omitempty"` + City string `json:"city"` + PostCode string `json:"post_code"` + Country string `json:"country"` // ISO 3166-1 alpha-2 +} + +// ── Sentinels ───────────────────────────────────────────────────────────── + +// ErrNoOptions is returned by a multi-provider orchestrator when no registered +// provider could serve the given address. A single provider returns an empty +// slice (not this error) for an unsupported address. +var ErrNoOptions = errors.New("shipping: no delivery options available") + +// ErrProviderUnavailable is returned when a provider's API is unreachable or +// returns a non-retryable error. The caller may fall back to another provider +// or degrade gracefully. +var ErrProviderUnavailable = errors.New("shipping: provider unavailable") + +// ── Wire helpers ────────────────────────────────────────────────────────── + +// Encode serialises a DeliveryOptionGroup slice for caching or event payloads. +func Encode(groups []DeliveryOptionGroup) ([]byte, error) { + return json.Marshal(groups) +} + +// Decode parses a DeliveryOptionGroup slice from Encode output. +func Decode(raw []byte) ([]DeliveryOptionGroup, error) { + var out []DeliveryOptionGroup + err := json.Unmarshal(raw, &out) + return out, err +} diff --git a/shipping/shipping_test.go b/shipping/shipping_test.go new file mode 100644 index 0000000..f65e45e --- /dev/null +++ b/shipping/shipping_test.go @@ -0,0 +1,306 @@ +package shipping + +import ( + "context" + "errors" + "sync" + "testing" +) + +// InMemoryProvider is a thread-safe in-memory implementation of Provider for +// tests — no external API calls needed. Backed by a map from +// "country|postcode" to available delivery option groups. +type InMemoryProvider struct { + mu sync.Mutex + opts map[string][]DeliveryOptionGroup // key: "country|postcode" + err error // injected error for testing failure paths +} + +// NewInMemoryProvider creates a provider seeded with delivery option groups. +// seed is a sequence of (country, postcode, []DeliveryOptionGroup) triples. +func NewInMemoryProvider(seed ...any) *InMemoryProvider { + p := &InMemoryProvider{ + opts: make(map[string][]DeliveryOptionGroup), + } + for i := 0; i+2 < len(seed); i += 3 { + country, ok := seed[i].(string) + if !ok { + continue + } + postcode, ok := seed[i+1].(string) + if !ok { + continue + } + groups, ok := seed[i+2].([]DeliveryOptionGroup) + if !ok { + continue + } + p.opts[providerKey(country, postcode)] = groups + } + return p +} + +func providerKey(country, postcode string) string { + return country + "|" + postcode +} + +// SetError injects an error that GetShippingOptions returns on every call +// (for testing failure paths). Nil clears the error. +func (p *InMemoryProvider) SetError(err error) { + p.mu.Lock() + defer p.mu.Unlock() + p.err = err +} + +func (p *InMemoryProvider) GetShippingOptions(_ context.Context, req GetOptionsRequest) ([]DeliveryOptionGroup, error) { + p.mu.Lock() + defer p.mu.Unlock() + + if p.err != nil { + return nil, p.err + } + + groups, ok := p.opts[providerKey(req.Country, req.PostCode)] + if !ok { + return []DeliveryOptionGroup{}, nil + } + return groups, nil +} + +// ── Tests ───────────────────────────────────────────────────────────────── + +func TestProviderReturnsOptions(t *testing.T) { + ctx := context.Background() + p := NewInMemoryProvider( + "SE", "17173", []DeliveryOptionGroup{ + { + Type: "home", + Name: "Home delivery", + DefaultOption: DeliveryOption{ + ID: "home-1", + Name: "Home delivery", + Description: "Delivered to your door", + DeliveryDays: "2-3 working days", + }, + }, + { + Type: "pickup", + Name: "Pickup point", + DefaultOption: DeliveryOption{ + ID: "pickup-1", + Name: "PostNord Storgatan", + Description: "Collect at PostNord Storgatan", + DeliveryDays: "2-3 working days", + PickupLocation: &PickupLocation{ + ID: "loc-1", + Name: "PostNord Storgatan", + Address: Address{ + Street: "Storgatan 9", + City: "Solna", + PostCode: "17173", + Country: "SE", + }, + }, + }, + }, + }, + ) + + groups, err := p.GetShippingOptions(ctx, GetOptionsRequest{ + Country: "SE", + PostCode: "17173", + Locale: "sv", + }) + if err != nil { + t.Fatalf("GetShippingOptions: %v", err) + } + if len(groups) != 2 { + t.Fatalf("expected 2 option groups, got %d", len(groups)) + } + + // Verify home delivery option + if groups[0].Type != "home" { + t.Fatalf("first group: expected type=home, got %q", groups[0].Type) + } + if groups[0].DefaultOption.ID != "home-1" { + t.Fatalf("home default option: expected ID=home-1, got %q", groups[0].DefaultOption.ID) + } + + // Verify pickup option with location + if groups[1].Type != "pickup" { + t.Fatalf("second group: expected type=pickup, got %q", groups[1].Type) + } + loc := groups[1].DefaultOption.PickupLocation + if loc == nil { + t.Fatal("pickup option: expected a PickupLocation, got nil") + } + if loc.Address.City != "Solna" { + t.Fatalf("pickup address city: expected Solna, got %q", loc.Address.City) + } +} + +func TestProviderUnsupportedAddress(t *testing.T) { + ctx := context.Background() + p := NewInMemoryProvider() // no seed data + + groups, err := p.GetShippingOptions(ctx, GetOptionsRequest{ + Country: "NO", + PostCode: "0150", + }) + if err != nil { + t.Fatalf("expected nil error for unsupported address, got %v", err) + } + if groups == nil { + t.Fatal("expected non-nil empty slice for unsupported address, got nil") + } + if len(groups) != 0 { + t.Fatalf("expected 0 option groups for unsupported address, got %d", len(groups)) + } +} + +func TestProviderError(t *testing.T) { + ctx := context.Background() + p := NewInMemoryProvider( + "SE", "17173", []DeliveryOptionGroup{ + { + Type: "home", + Name: "Home delivery", + DefaultOption: DeliveryOption{ + ID: "home-1", + Name: "Home delivery", + }, + }, + }, + ) + + injected := errors.New("api timeout") + p.SetError(injected) + + _, err := p.GetShippingOptions(ctx, GetOptionsRequest{ + Country: "SE", + PostCode: "17173", + }) + if err == nil { + t.Fatal("expected injected error, got nil") + } + if !errors.Is(err, injected) { + t.Fatalf("expected injected error %v, got %v", injected, err) + } +} + +func TestProviderAdditionalOptions(t *testing.T) { + ctx := context.Background() + + additionalPoints := make([]DeliveryOption, 2) + additionalPoints[0] = DeliveryOption{ + ID: "pickup-2", + Name: "ICA Maxi Solna", + PickupLocation: &PickupLocation{ + ID: "loc-2", + Name: "ICA Maxi Solna", + Address: Address{ + Street: "Solna Centrum 5", + City: "Solna", + PostCode: "17173", + Country: "SE", + }, + }, + } + additionalPoints[1] = DeliveryOption{ + ID: "pickup-3", + Name: "Coop Solna", + PickupLocation: &PickupLocation{ + ID: "loc-3", + Name: "Coop Solna", + Address: Address{ + Street: "Solna Torg 2", + City: "Solna", + PostCode: "17173", + Country: "SE", + }, + }, + } + + p := NewInMemoryProvider( + "SE", "17173", []DeliveryOptionGroup{ + { + Type: "pickup", + Name: "Pickup point", + DefaultOption: DeliveryOption{ + ID: "pickup-1", + Name: "PostNord Storgatan", + }, + AdditionalOptions: additionalPoints, + }, + }, + ) + + groups, err := p.GetShippingOptions(ctx, GetOptionsRequest{ + Country: "SE", + PostCode: "17173", + }) + if err != nil { + t.Fatalf("GetShippingOptions: %v", err) + } + if len(groups) != 1 { + t.Fatalf("expected 1 option group, got %d", len(groups)) + } + + addl := groups[0].AdditionalOptions + if len(addl) != 2 { + t.Fatalf("expected 2 additional options, got %d", len(addl)) + } + if addl[0].ID != "pickup-2" { + t.Fatalf("additional[0]: expected ID=pickup-2, got %q", addl[0].ID) + } +} + +func TestRoundTripEncodeDecode(t *testing.T) { + original := []DeliveryOptionGroup{ + { + Type: "home", + Name: "Home delivery", + DefaultOption: DeliveryOption{ + ID: "home-1", + Name: "Home delivery", + DeliveryDays: "2-3 working days", + }, + }, + } + + encoded, err := Encode(original) + if err != nil { + t.Fatalf("Encode: %v", err) + } + + decoded, err := Decode(encoded) + if err != nil { + t.Fatalf("Decode: %v", err) + } + + if len(decoded) != 1 { + t.Fatalf("round-trip: expected 1 group, got %d", len(decoded)) + } + if decoded[0].DefaultOption.ID != "home-1" { + t.Fatalf("round-trip: expected ID=home-1, got %q", decoded[0].DefaultOption.ID) + } +} + +func TestNoOptionsSentinelNotReturnedBySingleProvider(t *testing.T) { + // ErrNoOptions is for multi-provider orchestrators. A single provider +// returns an empty slice for unsupported addresses. Verify that the +// in-memory provider follows this contract. + ctx := context.Background() + p := NewInMemoryProvider() + + groups, err := p.GetShippingOptions(ctx, GetOptionsRequest{ + Country: "FI", + PostCode: "00100", + }) + if err != nil { + t.Fatalf("expected nil for unsupported address, got %v", err) + } + if groups == nil { + t.Fatal("expected non-nil empty slice, got nil") + } +}