diff --git a/internal/lifecycle/lifecycle.go b/internal/lifecycle/lifecycle.go new file mode 100644 index 0000000..4a81cb5 --- /dev/null +++ b/internal/lifecycle/lifecycle.go @@ -0,0 +1,145 @@ +// Package lifecycle manages the status of stored bug-reports — pending, removed, +// or published (#10) — on top of the storage.ObjectStore seam. +// +// # Status model +// +// Status is encoded in the object key, not inside the (encrypted) object body: +// every report lives at reports// (see storage.ReportKey). A report +// is born pending at ingest (storage.Sink) and moves to exactly one terminal +// state: +// +// pending ──MarkRemoved──▶ removed (a maintainer pulled it, #11) +// pending ──MarkPublished▶ published (it became a GitHub issue, #15) +// +// Because status is the key prefix, "list the pending reports" (used by the +// weekly job, #13) is a single prefix listing with no secondary index that could +// drift out of sync, so it returns exactly the pending reports by construction. +// +// # Transitions move ciphertext only +// +// A transition copies the stored object — which is an opaque AES-256-GCM frame +// (internal/crypto) — to the destination status key and deletes the source key. +// The bytes are never decrypted, re-encrypted, or otherwise inspected; no key is +// needed to change status. This keeps the plaintext exposure of the pipeline +// unchanged and lets an unprivileged component (e.g. the maintainer removal tool) +// change status without the decryption key. +// +// # Atomicity, idempotency, retry-safety +// +// Object stores offer no multi-key transaction, so a transition is a copy (Put +// destination) followed by a delete (source), in that order. The ordering is +// chosen to never lose a report: if the process dies after the Put but before the +// Delete, the object is momentarily visible under *both* statuses, but no data is +// lost. Re-running the same transition converges — the Put rewrites identical +// bytes and the Delete removes the leftover source — so a transition is safe to +// retry unconditionally. Deleting an already-absent key is a no-op in every +// ObjectStore, so the retry does not spuriously fail. +// +// A transition whose source is absent is treated as idempotent success when the +// object is already at the destination (an earlier attempt completed), and as an +// ErrUnknownReport otherwise (the id was never pending, or is in a different +// terminal state). Callers therefore may retry MarkRemoved/MarkPublished freely. +package lifecycle + +import ( + "context" + "errors" + "fmt" + "strings" + + "github.com/JMR-dev/LibreMail-Bug-Report-Ingest/internal/storage" +) + +// ErrUnknownReport is returned by a transition when no report with the given id +// exists in the expected source status (and it is not already at the +// destination). It wraps nothing sensitive: the id is an opaque handle. +var ErrUnknownReport = errors.New("lifecycle: unknown pending report id") + +// Manager reads and transitions report status over an ObjectStore. It is +// stateless beyond the store reference and safe for concurrent use to the same +// extent the underlying store is. +type Manager struct { + store storage.ObjectStore +} + +// New returns a Manager backed by store. In production store is a *storage.R2Store +// bound to the same bucket the ingest Sink writes to; in tests and cmd/devserver +// it is a *storage.MemoryStore. +func New(store storage.ObjectStore) *Manager { + return &Manager{store: store} +} + +// ListPending returns the ids of all reports currently pending, in ascending +// order (the id's leading UTC timestamp makes this oldest-first, which is the +// order the weekly publish job wants). Given a mix of pending, removed, and +// published reports it returns exactly the pending ones — the removed and +// published reports live under different key prefixes and are not listed. +func (m *Manager) ListPending(ctx context.Context) ([]string, error) { + prefix := storage.StatusPrefix(storage.StatusPending) + keys, err := m.store.List(ctx, prefix) + if err != nil { + return nil, fmt.Errorf("lifecycle: list pending: %w", err) + } + ids := make([]string, 0, len(keys)) + for _, k := range keys { + ids = append(ids, strings.TrimPrefix(k, prefix)) + } + return ids, nil +} + +// GetPending returns the stored (still-encrypted) frame for a pending report, or +// storage.ErrNotFound if no such pending report exists. The publish job (#14/#15) +// uses it to fetch a report's ciphertext before decrypting and publishing it; +// this package does not decrypt. +func (m *Manager) GetPending(ctx context.Context, id string) ([]byte, error) { + return m.store.Get(ctx, storage.ReportKey(storage.StatusPending, id)) +} + +// MarkRemoved transitions a pending report to removed (#11), excluding it from +// future publish runs. It is idempotent: removing an already-removed report +// succeeds. Removing an id that is not pending (unknown, or already published) +// returns ErrUnknownReport. +func (m *Manager) MarkRemoved(ctx context.Context, id string) error { + return m.transition(ctx, id, storage.StatusPending, storage.StatusRemoved) +} + +// MarkPublished transitions a pending report to published (#15) after it has been +// posted as a GitHub issue, so the next run does not re-publish it. It is +// idempotent: publishing an already-published report succeeds. Publishing an id +// that is not pending (unknown, or already removed) returns ErrUnknownReport. +func (m *Manager) MarkPublished(ctx context.Context, id string) error { + return m.transition(ctx, id, storage.StatusPending, storage.StatusPublished) +} + +// transition moves report id from status to a new status by copying the encrypted +// object to the destination key and deleting the source key. See the package doc +// for the atomicity/idempotency/retry model. +func (m *Manager) transition(ctx context.Context, id string, from, to storage.Status) error { + fromKey := storage.ReportKey(from, id) + toKey := storage.ReportKey(to, id) + + data, err := m.store.Get(ctx, fromKey) + if errors.Is(err, storage.ErrNotFound) { + // Source absent. Either an earlier attempt already completed the move (the + // object is at the destination -> idempotent success), or the id was never + // in the source status (-> unknown). + if _, derr := m.store.Get(ctx, toKey); derr == nil { + return nil + } + return fmt.Errorf("%w: %q", ErrUnknownReport, id) + } + if err != nil { + return fmt.Errorf("lifecycle: read %s report: %w", from, err) + } + + // Copy the opaque ciphertext to the destination, then drop the source. Put + // before Delete so an interruption never loses the report; the operation is + // idempotent, so a retry converges. + if err := m.store.Put(ctx, toKey, data); err != nil { + return fmt.Errorf("lifecycle: write %s report: %w", to, err) + } + if err := m.store.Delete(ctx, fromKey); err != nil { + return fmt.Errorf("lifecycle: delete %s report: %w", from, err) + } + return nil +} diff --git a/internal/lifecycle/lifecycle_test.go b/internal/lifecycle/lifecycle_test.go new file mode 100644 index 0000000..fa3a72d --- /dev/null +++ b/internal/lifecycle/lifecycle_test.go @@ -0,0 +1,279 @@ +package lifecycle_test + +import ( + "bytes" + "context" + "errors" + "slices" + "testing" + + "github.com/JMR-dev/LibreMail-Bug-Report-Ingest/internal/crypto" + "github.com/JMR-dev/LibreMail-Bug-Report-Ingest/internal/lifecycle" + "github.com/JMR-dev/LibreMail-Bug-Report-Ingest/internal/scrub" + "github.com/JMR-dev/LibreMail-Bug-Report-Ingest/internal/storage" +) + +// mustKeyring builds a single-key keyring for sealing realistic ciphertext. +func mustKeyring(t *testing.T) *crypto.Keyring { + t.Helper() + key, err := crypto.GenerateKey() + if err != nil { + t.Fatalf("GenerateKey: %v", err) + } + kr, err := crypto.NewKeyring(1, map[uint16][]byte{1: key}) + if err != nil { + t.Fatalf("NewKeyring: %v", err) + } + return kr +} + +// seal returns an encrypted frame for plaintext (a realistic stored object). +func seal(t *testing.T, kr *crypto.Keyring, plaintext string) []byte { + t.Helper() + frame, err := crypto.Seal(kr, []byte(plaintext)) + if err != nil { + t.Fatalf("Seal: %v", err) + } + return frame +} + +// putStatus writes frame directly at the reports// key, seeding a +// report in a chosen status without going through a transition. +func putStatus(t *testing.T, store storage.ObjectStore, status storage.Status, id string, frame []byte) { + t.Helper() + if err := store.Put(context.Background(), storage.ReportKey(status, id), frame); err != nil { + t.Fatalf("seed %s/%s: %v", status, id, err) + } +} + +// TestListPendingExactness is the ticket's acceptance criterion: given a mix of +// reports in different states, ListPending returns exactly the pending ids. +func TestListPendingExactness(t *testing.T) { + ctx := context.Background() + kr := mustKeyring(t) + store := storage.NewMemoryStore() + m := lifecycle.New(store) + + // A deliberately interleaved mix across all three states. + putStatus(t, store, storage.StatusPending, "id-a", seal(t, kr, "a")) + putStatus(t, store, storage.StatusRemoved, "id-b", seal(t, kr, "b")) + putStatus(t, store, storage.StatusPending, "id-c", seal(t, kr, "c")) + putStatus(t, store, storage.StatusPublished, "id-d", seal(t, kr, "d")) + putStatus(t, store, storage.StatusPending, "id-e", seal(t, kr, "e")) + putStatus(t, store, storage.StatusRemoved, "id-f", seal(t, kr, "f")) + + got, err := m.ListPending(ctx) + if err != nil { + t.Fatalf("ListPending: %v", err) + } + want := []string{"id-a", "id-c", "id-e"} // sorted; only the pending ones + if !slices.Equal(got, want) { + t.Errorf("ListPending = %v, want exactly %v", got, want) + } +} + +// TestNewReportsAreListedAsPending proves the #9 ingest Sink now lands reports in +// the pending state, discoverable via ListPending, and readable back as the same +// ciphertext (which still decrypts to the scrubbed body). +func TestNewReportsAreListedAsPending(t *testing.T) { + ctx := context.Background() + kr := mustKeyring(t) + store := storage.NewMemoryStore() + sink := storage.NewSink(store, kr) // default key func -> reports/pending/ + m := lifecycle.New(store) + + raw := `{"appVersion":"1.0.0","platform":"android","report":"boom from 203.0.113.7"}` + if err := sink.Store(ctx, []byte(raw)); err != nil { + t.Fatalf("sink.Store: %v", err) + } + + ids, err := m.ListPending(ctx) + if err != nil { + t.Fatalf("ListPending: %v", err) + } + if len(ids) != 1 { + t.Fatalf("ListPending returned %d ids, want 1 (the just-ingested report)", len(ids)) + } + + frame, err := m.GetPending(ctx, ids[0]) + if err != nil { + t.Fatalf("GetPending(%q): %v", ids[0], err) + } + opened, err := crypto.Open(kr, frame) + if err != nil { + t.Fatalf("Open pending frame: %v", err) + } + if !bytes.Equal(opened, scrub.Scrub([]byte(raw))) { + t.Error("pending frame did not decrypt to the scrubbed body") + } +} + +// TestMarkRemoved checks pending -> removed: the report leaves the pending set, +// appears under removed, and its bytes are the unchanged ciphertext. +func TestMarkRemoved(t *testing.T) { + transitionCase(t, "removed", + func(m *lifecycle.Manager, id string) error { return m.MarkRemoved(context.Background(), id) }, + storage.StatusRemoved) +} + +// TestMarkPublished checks pending -> published with the same guarantees. +func TestMarkPublished(t *testing.T) { + transitionCase(t, "published", + func(m *lifecycle.Manager, id string) error { return m.MarkPublished(context.Background(), id) }, + storage.StatusPublished) +} + +// transitionCase is the shared body for the two happy-path transitions. +func transitionCase(t *testing.T, name string, do func(*lifecycle.Manager, string) error, target storage.Status) { + t.Helper() + ctx := context.Background() + kr := mustKeyring(t) + store := storage.NewMemoryStore() + m := lifecycle.New(store) + + const id = "id-x" + frame := seal(t, kr, "sensitive report body") + putStatus(t, store, storage.StatusPending, id, frame) + // A second pending report that must be unaffected by the transition. + putStatus(t, store, storage.StatusPending, "id-keep", seal(t, kr, "other")) + + if err := do(m, id); err != nil { + t.Fatalf("%s transition: %v", name, err) + } + + // It left the pending set (only the untouched report remains). + pending, _ := m.ListPending(ctx) + if !slices.Equal(pending, []string{"id-keep"}) { + t.Errorf("after %s, ListPending = %v, want [id-keep]", name, pending) + } + + // The source key is gone. + if _, err := store.Get(ctx, storage.ReportKey(storage.StatusPending, id)); !errors.Is(err, storage.ErrNotFound) { + t.Errorf("pending key still present after %s (err=%v)", name, err) + } + + // It appears under the target status with byte-identical ciphertext... + moved, err := store.Get(ctx, storage.ReportKey(target, id)) + if err != nil { + t.Fatalf("get %s object: %v", target, err) + } + if !bytes.Equal(moved, frame) { + t.Errorf("%s object bytes changed by the transition; must move ciphertext verbatim", target) + } + // ...and the frame is intact: it still decrypts to the original plaintext, + // proving the transition neither decrypted nor corrupted it. + opened, err := crypto.Open(kr, moved) + if err != nil { + t.Fatalf("Open moved frame: %v", err) + } + if string(opened) != "sensitive report body" { + t.Errorf("moved frame decrypted to %q, want original plaintext", opened) + } +} + +// TestTransitionIdempotentRetry proves a transition can be retried safely: calling +// it twice ends in the same single-object state and the second call returns nil. +func TestTransitionIdempotentRetry(t *testing.T) { + ctx := context.Background() + kr := mustKeyring(t) + store := storage.NewMemoryStore() + m := lifecycle.New(store) + + const id = "id-1" + putStatus(t, store, storage.StatusPending, id, seal(t, kr, "body")) + + if err := m.MarkPublished(ctx, id); err != nil { + t.Fatalf("first MarkPublished: %v", err) + } + if err := m.MarkPublished(ctx, id); err != nil { + t.Errorf("second MarkPublished (retry) = %v, want nil (idempotent)", err) + } + + if pending, _ := m.ListPending(ctx); len(pending) != 0 { + t.Errorf("ListPending = %v after publish, want empty", pending) + } + if published, _ := store.List(ctx, storage.StatusPrefix(storage.StatusPublished)); len(published) != 1 { + t.Errorf("published objects = %d, want exactly 1 (no duplicate from retry)", len(published)) + } +} + +// TestTransitionConvergesFromInterruptedState simulates a crash after the copy +// (Put destination) but before the delete (source): the object exists under both +// statuses. Re-running the transition must converge to only the destination. +func TestTransitionConvergesFromInterruptedState(t *testing.T) { + ctx := context.Background() + kr := mustKeyring(t) + store := storage.NewMemoryStore() + m := lifecycle.New(store) + + const id = "id-2" + frame := seal(t, kr, "body") + putStatus(t, store, storage.StatusPending, id, frame) // source (leftover) + putStatus(t, store, storage.StatusPublished, id, frame) // destination (already copied) + + // Mid-transition the report is visible in both sets. + if pending, _ := m.ListPending(ctx); !slices.Contains(pending, id) { + t.Fatalf("precondition: expected %q to appear pending mid-transition", id) + } + + if err := m.MarkPublished(ctx, id); err != nil { + t.Fatalf("MarkPublished (convergence): %v", err) + } + + if pending, _ := m.ListPending(ctx); slices.Contains(pending, id) { + t.Errorf("after convergence, %q still pending", id) + } + if _, err := store.Get(ctx, storage.ReportKey(storage.StatusPublished, id)); err != nil { + t.Errorf("published object missing after convergence: %v", err) + } +} + +// TestUnknownIdErrors covers transitions of ids that are not pending. +func TestUnknownIdErrors(t *testing.T) { + ctx := context.Background() + kr := mustKeyring(t) + store := storage.NewMemoryStore() + m := lifecycle.New(store) + + // Never-seen id. + if err := m.MarkRemoved(ctx, "ghost"); !errors.Is(err, lifecycle.ErrUnknownReport) { + t.Errorf("MarkRemoved(unknown) err = %v, want ErrUnknownReport", err) + } + if err := m.MarkPublished(ctx, "ghost"); !errors.Is(err, lifecycle.ErrUnknownReport) { + t.Errorf("MarkPublished(unknown) err = %v, want ErrUnknownReport", err) + } + + // An id in a terminal state cannot be transitioned again from pending: a + // removed report is not publishable, and a published report is not removable. + putStatus(t, store, storage.StatusRemoved, "was-removed", seal(t, kr, "x")) + if err := m.MarkPublished(ctx, "was-removed"); !errors.Is(err, lifecycle.ErrUnknownReport) { + t.Errorf("MarkPublished(removed id) err = %v, want ErrUnknownReport", err) + } + putStatus(t, store, storage.StatusPublished, "was-published", seal(t, kr, "y")) + if err := m.MarkRemoved(ctx, "was-published"); !errors.Is(err, lifecycle.ErrUnknownReport) { + t.Errorf("MarkRemoved(published id) err = %v, want ErrUnknownReport", err) + } +} + +// TestGetPending covers the report-fetch helper the publish job relies on. +func TestGetPending(t *testing.T) { + ctx := context.Background() + kr := mustKeyring(t) + store := storage.NewMemoryStore() + m := lifecycle.New(store) + + frame := seal(t, kr, "body") + putStatus(t, store, storage.StatusPending, "id-1", frame) + + got, err := m.GetPending(ctx, "id-1") + if err != nil { + t.Fatalf("GetPending: %v", err) + } + if !bytes.Equal(got, frame) { + t.Error("GetPending returned different bytes than were stored") + } + + if _, err := m.GetPending(ctx, "missing"); !errors.Is(err, storage.ErrNotFound) { + t.Errorf("GetPending(missing) err = %v, want ErrNotFound", err) + } +} diff --git a/internal/storage/memory.go b/internal/storage/memory.go index a9c1038..7e08865 100644 --- a/internal/storage/memory.go +++ b/internal/storage/memory.go @@ -3,6 +3,7 @@ package storage import ( "context" "sort" + "strings" "sync" ) @@ -39,6 +40,30 @@ func (m *MemoryStore) Get(_ context.Context, key string) ([]byte, error) { return append([]byte(nil), data...), nil } +// List returns the stored keys that start with prefix, in sorted order. A prefix +// of "" lists every key (equivalent to Keys). +func (m *MemoryStore) List(_ context.Context, prefix string) ([]string, error) { + m.mu.Lock() + defer m.mu.Unlock() + var keys []string + for k := range m.objects { + if strings.HasPrefix(k, prefix) { + keys = append(keys, k) + } + } + sort.Strings(keys) + return keys, nil +} + +// Delete removes the object at key. Deleting an absent key is a no-op (nil), +// matching R2's idempotent delete so a retried transition behaves identically. +func (m *MemoryStore) Delete(_ context.Context, key string) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.objects, key) + return nil +} + // Keys returns the stored object keys in sorted order (test/ops helper). func (m *MemoryStore) Keys() []string { m.mu.Lock() diff --git a/internal/storage/r2_wasm.go b/internal/storage/r2_wasm.go index 0bf4fac..e0de44d 100644 --- a/internal/storage/r2_wasm.go +++ b/internal/storage/r2_wasm.go @@ -12,13 +12,19 @@ import ( "bytes" "context" "io" + "syscall/js" + "github.com/syumai/workers/cloudflare" "github.com/syumai/workers/cloudflare/r2" ) // R2Store is an ObjectStore backed by a Cloudflare R2 bucket binding. type R2Store struct { bucket *r2.Bucket + // binding is the raw R2 binding value. The syumai r2.Bucket.List helper takes + // no options (no prefix, no cursor), so List drives the underlying JS list() + // directly to page a prefix — see List. + binding js.Value } // NewR2Store resolves the R2 bucket bound as binding (see BucketBinding) from the @@ -29,7 +35,7 @@ func NewR2Store(binding string) (*R2Store, error) { if err != nil { return nil, err } - return &R2Store{bucket: b}, nil + return &R2Store{bucket: b, binding: cloudflare.GetBinding(binding)}, nil } // Put writes data at key. The syumai R2 Put reads the whole body into memory and @@ -51,4 +57,44 @@ func (s *R2Store) Get(_ context.Context, key string) ([]byte, error) { return io.ReadAll(obj.Body) } +// List returns every key that starts with prefix, in the ascending order R2 +// returns them (list is lexicographically ordered). It pages through the whole +// result set via the cursor so a status with more than one list page (R2 caps a +// page at 1000) is still enumerated exactly — important for "list pending" to be +// complete. Only keys are read; object bodies (ciphertext) are never fetched. +func (s *R2Store) List(_ context.Context, prefix string) ([]string, error) { + var keys []string + cursor := "" + for { + // Build the list options object with Object/Set (the idiom the syumai R2 + // helpers use) rather than js.ValueOf(map), which keeps the JS-interop + // surface identical to the rest of the wasm build. + opts := js.Global().Get("Object").New() + opts.Set("prefix", prefix) + opts.Set("limit", 1000) + if cursor != "" { + opts.Set("cursor", cursor) + } + res, err := await(s.binding.Call("list", opts)) + if err != nil { + return nil, err + } + objs := res.Get("objects") + for i := 0; i < objs.Length(); i++ { + keys = append(keys, objs.Index(i).Get("key").String()) + } + if !res.Get("truncated").Bool() { + break + } + cursor = res.Get("cursor").String() + } + return keys, nil +} + +// Delete removes the object at key. R2's delete is idempotent (deleting an absent +// key succeeds), which is what makes a retried status transition safe. +func (s *R2Store) Delete(_ context.Context, key string) error { + return s.bucket.Delete(key) +} + var _ ObjectStore = (*R2Store)(nil) diff --git a/internal/storage/sink.go b/internal/storage/sink.go index 0bf008d..793d29d 100644 --- a/internal/storage/sink.go +++ b/internal/storage/sink.go @@ -12,9 +12,6 @@ import ( "github.com/JMR-dev/LibreMail-Bug-Report-Ingest/internal/scrub" ) -// objectKeyPrefix namespaces report objects within the bucket. -const objectKeyPrefix = "reports/" - // Sink is the real ingest.Sink. For each accepted report it: // // 1. scrubs the raw body (best-effort PII redaction, #8); @@ -65,14 +62,22 @@ func (s *Sink) Store(ctx context.Context, raw []byte) error { return nil } -// defaultObjectKey builds a unique object key: a UTC timestamp (for rough +// defaultObjectKey builds the key for a freshly accepted report. Reports enter +// the lifecycle (#10) as pending, so the key lives under the pending status +// prefix: reports/pending/. The id is a UTC timestamp (for rough // lexicographic ordering, convenient for the weekly publish job) plus 80 bits of -// CSPRNG randomness (so keys are unguessable and collision-free within a second). +// CSPRNG randomness (so ids are unguessable and collision-free within a second), +// and is stable as the report later transitions to published/removed. func defaultObjectKey() string { + return ReportKey(StatusPending, newReportID()) +} + +// newReportID mints a stable, unguessable report id: -<80-bit-rand>. +func newReportID() string { var b [10]byte _, _ = rand.Read(b[:]) ts := time.Now().UTC().Format("20060102T150405") - return objectKeyPrefix + ts + "-" + hex.EncodeToString(b[:]) + return ts + "-" + hex.EncodeToString(b[:]) } // Sink satisfies the ingest storage seam. diff --git a/internal/storage/status.go b/internal/storage/status.go new file mode 100644 index 0000000..82a5a58 --- /dev/null +++ b/internal/storage/status.go @@ -0,0 +1,50 @@ +package storage + +// Report lifecycle status, encoded in the object-key layout (#10). +// +// Every stored report lives under a status-partitioned key: +// +// reports// +// +// where is the stable, unguessable report id minted at ingest +// (see defaultObjectKey) and is one of the values below. Encoding the +// status in the key prefix means: +// +// - "list the pending reports" is a single prefix List(StatusPrefix(StatusPending)), +// with no separate index to keep in sync; and +// - a status transition is a copy of the (already-encrypted) object to the new +// status's key followed by a delete of the old key. The stored bytes are pure +// ciphertext and are never decrypted, re-encrypted, or otherwise touched to +// change status (see internal/lifecycle). +// +// The is stable across transitions: only the status segment of the key +// changes, so a report keeps its identity as it moves pending -> published/removed. +type Status string + +const ( + // StatusPending is a freshly accepted report awaiting the weekly publish job. + // New reports are written here by the ingest Sink. + StatusPending Status = "pending" + // StatusRemoved is a report a maintainer pulled before publication (#11); it + // is excluded from publishing. + StatusRemoved Status = "removed" + // StatusPublished is a report already published as a GitHub issue (#15); it is + // retained but never re-published. + StatusPublished Status = "published" +) + +// ReportsPrefix namespaces every report object within the bucket. +const ReportsPrefix = "reports/" + +// StatusPrefix is the key prefix shared by all reports in status s, e.g. +// "reports/pending/". Listing this prefix enumerates exactly the reports in that +// status. The trailing slash is load-bearing: it keeps the statuses disjoint so +// (for example) "reports/pending/" never matches a "reports/published/..." key. +func StatusPrefix(s Status) string { + return ReportsPrefix + string(s) + "/" +} + +// ReportKey is the object key for report id in status s: reports//. +func ReportKey(s Status, id string) string { + return StatusPrefix(s) + id +} diff --git a/internal/storage/storage_test.go b/internal/storage/storage_test.go index e94468d..c575e6a 100644 --- a/internal/storage/storage_test.go +++ b/internal/storage/storage_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "slices" "strings" "testing" @@ -86,13 +87,14 @@ func TestSinkFullPath(t *testing.T) { t.Fatalf("Store: %v", err) } - // Exactly one object, under a reports/ key. + // Exactly one object, and new reports enter the lifecycle as pending, so the + // key lives under the pending status prefix (#10). keys := store.Keys() if len(keys) != 1 { t.Fatalf("stored %d objects, want 1", len(keys)) } - if !strings.HasPrefix(keys[0], objectKeyPrefix) { - t.Errorf("object key %q lacks prefix %q", keys[0], objectKeyPrefix) + if want := StatusPrefix(StatusPending); !strings.HasPrefix(keys[0], want) { + t.Errorf("object key %q lacks pending prefix %q", keys[0], want) } stored, err := store.Get(ctx, keys[0]) @@ -164,6 +166,12 @@ func (failingStore) Put(context.Context, string, []byte) error { func (failingStore) Get(context.Context, string) ([]byte, error) { return nil, ErrNotFound } +func (failingStore) List(context.Context, string) ([]string, error) { + return nil, errors.New("backend down") +} +func (failingStore) Delete(context.Context, string) error { + return errors.New("backend down") +} func TestSinkStorePutError(t *testing.T) { sink := NewSink(failingStore{}, mustKeyring(t)) @@ -197,3 +205,55 @@ func TestSinkWithKeyFunc(t *testing.T) { t.Errorf("expected object at fixed key: %v", err) } } + +// TestStatusKeyLayout pins the status key scheme the lifecycle layer relies on: +// disjoint, trailing-slashed status prefixes so a prefix List of one status can +// never match another. +func TestStatusKeyLayout(t *testing.T) { + if got := ReportKey(StatusPending, "abc"); got != "reports/pending/abc" { + t.Errorf("ReportKey(pending, abc) = %q, want reports/pending/abc", got) + } + // "reports/pending/" must not be a prefix of another status's keys. + if strings.HasPrefix(ReportKey(StatusPublished, "x"), StatusPrefix(StatusPending)) { + t.Error("pending prefix wrongly matches a published key") + } + if strings.HasPrefix(ReportKey(StatusRemoved, "x"), StatusPrefix(StatusPending)) { + t.Error("pending prefix wrongly matches a removed key") + } +} + +func TestMemoryStoreListAndDelete(t *testing.T) { + ctx := context.Background() + m := NewMemoryStore() + seed := map[string][]byte{ + "reports/pending/a": []byte("1"), + "reports/pending/b": []byte("2"), + "reports/published/c": []byte("3"), + "reports/removed/d": []byte("4"), + } + for k, v := range seed { + if err := m.Put(ctx, k, v); err != nil { + t.Fatalf("Put %s: %v", k, err) + } + } + + got, err := m.List(ctx, "reports/pending/") + if err != nil { + t.Fatalf("List: %v", err) + } + want := []string{"reports/pending/a", "reports/pending/b"} + if !slices.Equal(got, want) { + t.Errorf("List(pending) = %v, want %v (must be exact and sorted)", got, want) + } + + // Delete is idempotent: deleting a present then an absent key both succeed. + if err := m.Delete(ctx, "reports/pending/a"); err != nil { + t.Fatalf("Delete present: %v", err) + } + if err := m.Delete(ctx, "reports/pending/a"); err != nil { + t.Fatalf("Delete absent (must be a no-op): %v", err) + } + if got, _ := m.List(ctx, "reports/pending/"); !slices.Equal(got, []string{"reports/pending/b"}) { + t.Errorf("after delete, List(pending) = %v, want [reports/pending/b]", got) + } +} diff --git a/internal/storage/store.go b/internal/storage/store.go index 4ef01c0..67a3376 100644 --- a/internal/storage/store.go +++ b/internal/storage/store.go @@ -37,9 +37,16 @@ const ( // ever handle already-encrypted frames. // // - Put writes data (a sealed frame) at key, overwriting any existing object. -// - Get reads the bytes back, or returns ErrNotFound. Get exists mainly for the -// future publish job (#35) and for tests; the ingest path is write-only. +// - Get reads the bytes back, or returns ErrNotFound. Get exists for the publish +// job (#15) and for tests; the ingest path is write-only. +// - List returns the keys whose names start with prefix, in sorted (ascending) +// order. It backs the lifecycle "list pending" query (#10/#13); implementations +// must page through the whole result set, not just the first batch. +// - Delete removes the object at key. Deleting a key that does not exist is not +// an error (idempotent), which is what makes a status transition safe to retry. type ObjectStore interface { Put(ctx context.Context, key string, data []byte) error Get(ctx context.Context, key string) ([]byte, error) + List(ctx context.Context, prefix string) ([]string, error) + Delete(ctx context.Context, key string) error } diff --git a/internal/storage/worker_sink_wasm.go b/internal/storage/worker_sink_wasm.go index 2b33165..f80bd96 100644 --- a/internal/storage/worker_sink_wasm.go +++ b/internal/storage/worker_sink_wasm.go @@ -96,7 +96,8 @@ func getSecret(binding string) ([]byte, error) { } // await resolves a JS Promise from the calling goroutine (the Worker runs each -// handler in its own goroutine, so the JS event loop can settle the promise). +// handler in its own goroutine, so the JS event loop can settle the promise). It +// is shared by the Secrets Store read (getSecret) and the R2 list paging (List). func await(p js.Value) (js.Value, error) { resCh := make(chan js.Value, 1) errCh := make(chan error, 1) @@ -118,7 +119,7 @@ func await(p js.Value) (js.Value, error) { if len(args) > 0 { msg = args[0].Call("toString").String() } - errCh <- fmt.Errorf("secrets store: %s", msg) + errCh <- fmt.Errorf("await: promise rejected: %s", msg) return js.Undefined() }) p.Call("then", then).Call("catch", catch)