From 21a725ff78a52380c384f99a51ebe0ad5a466984 Mon Sep 17 00:00:00 2001 From: Sheng Kun Chang Date: Wed, 17 Jun 2026 16:08:36 +0800 Subject: [PATCH] feat: meter transport -> dispatch-HTTP (drop Lambda) Replace the usage-meter transport with a dispatch-HTTP POST, modeled exactly on ms.Emit. ms.Record now POSTs the Event envelope to the platform dispatch usage ingress at {MS_DISPATCH_URL | dev fallback}/ apps/{appID}/usage in both dev and prod; dispatch re-derives the authoritative app/module/owner/recorded_at and forwards to billing-engine, so the SDK Hint fields stay untrusted. - Remove MS_METER_LAMBDA_ARN, lambda.Invoke, NewFromARN/NewDev, arnPattern, lambdaInvoker, and the aws-sdk-go-v2/service/lambda dependency. A single meter.New() builds the always-present HTTP client and fail-fast validates MS_DISPATCH_URL when set. - appID comes from the request context's auth identity; an empty appID returns an error (no panic, no HTTP call), mirroring ms.Emit. - EventID is still minted once per Record and reused across any transport retry so ON CONFLICT(event_id) dedupes. - Public API unchanged: ms.Meter declaration + ms.Record by-name, envelope v1 with no kind on the wire, reserved-name guards intact. - Tests rewritten onto an httptest dispatch stub (posts to /apps/{appID}/usage, retry reuses eventId, empty-app errors without HTTP, dev fallback URL, non-2xx surfaces body); ARN/Lambda tests removed. - VERSION 0.2.3 -> 0.2.4 + CHANGELOG entry. Co-Authored-By: Claude Opus 4.8 (1M context) --- CHANGELOG.md | 10 ++ VERSION | 2 +- examples/template/go.mod | 1 - examples/template/go.sum | 2 - go.mod | 1 - go.sum | 2 - internal/core/meter_test.go | 27 ++- internal/core/module.go | 23 ++- meter/meter.go | 179 +++++++++++-------- meter/meter_test.go | 345 +++++++++++++++++++----------------- 10 files changed, 337 insertions(+), 255 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 150ba34..8196a17 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,16 @@ and this project uses [Semantic Versioning](https://semver.org/spec/v2.0.0.html) ## [Unreleased] +## [v0.2.4] - 2026-06-17 + +The usage-meter transport moves from an AWS Lambda invoke to a dispatch-HTTP POST, exactly mirroring `ms.Emit`. The public metering API (`ms.Meter` declaration, `ms.Record` emit-by-name, the v1 envelope with no kind on the wire, the reserved-namespace guards) is unchanged — only how a recorded event reaches the platform changes. + +### Changed +- **Meter transport is now dispatch-HTTP, not Lambda.** `ms.Record` POSTs the usage `Event` envelope to the platform dispatch usage ingress at `{MS_DISPATCH_URL | dev fallback}/apps/{appID}/usage` (the same transport idiom as `ms.Emit` / `ms.Call`), in **both dev and prod** — there is no separate dev sink. Dispatch re-derives the authoritative app/module/owner/recorded_at and forwards to billing-engine; the SDK's `Hint` fields stay untrusted. The app id comes from the request context's auth identity (`auth.Get`), and an empty app id returns an error (no panic, no HTTP call), mirroring `ms.Emit`. The transport HTTP client is built once at `meter.New()` and is never nil. The non-fatal contract is unchanged: a transport failure is returned (then logged by the caller), never propagated to fail the handler. The `EventID` is still minted once per `Record` call and reused across any transport retry, so the platform's `ON CONFLICT(event_id)` dedupe holds. + +### Removed +- **`MS_METER_LAMBDA_ARN` and the AWS Lambda meter transport.** The `meter` package no longer invokes a Lambda: the `MS_METER_LAMBDA_ARN` environment variable is gone, along with `meter.NewFromARN` / `meter.NewDev` (replaced by a single `meter.New()`), the ARN-format validation, and the `github.com/aws/aws-sdk-go-v2/service/lambda` dependency. Usage transport is configured via **`MS_DISPATCH_URL`** (the same base `ms.Emit` / `ms.Call` use); `meter.New()` fail-fast validates it as a parseable http(s) base when set, so a typo surfaces at startup rather than as silently lost usage. An unset `MS_DISPATCH_URL` falls back to the local dispatch for dev. + ## [v0.2.3] - 2026-06-17 Metric kind moves from a positional argument to a declaration OPTION, and a module may now override the **customer** price of a platform-infra metric. The runtime emit (`ms.Record`) and the declaration-first contract are unchanged; only the `ms.Meter` shape and the reserved-namespace rules change. diff --git a/VERSION b/VERSION index 7179039..abd4105 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.2.3 +0.2.4 diff --git a/examples/template/go.mod b/examples/template/go.mod index a3eccd8..1e1b343 100644 --- a/examples/template/go.mod +++ b/examples/template/go.mod @@ -23,7 +23,6 @@ require ( github.com/aws/aws-sdk-go-v2/service/internal/checksum v1.9.13 // indirect github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21 // indirect github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21 // indirect - github.com/aws/aws-sdk-go-v2/service/lambda v1.89.0 // indirect github.com/aws/aws-sdk-go-v2/service/s3 v1.98.0 // indirect github.com/aws/aws-sdk-go-v2/service/signin v1.0.9 // indirect github.com/aws/aws-sdk-go-v2/service/sqs v1.42.25 // indirect diff --git a/examples/template/go.sum b/examples/template/go.sum index 9fea993..d12e7d7 100644 --- a/examples/template/go.sum +++ b/examples/template/go.sum @@ -28,8 +28,6 @@ github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21 h1:c31//R3x github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21/go.mod h1:r6+pf23ouCB718FUxaqzZdbpYFyDtehyZcmP5KL9FkA= github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21 h1:ZlvrNcHSFFWURB8avufQq9gFsheUgjVD9536obIknfM= github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21/go.mod h1:cv3TNhVrssKR0O/xxLJVRfd2oazSnZnkUeTf6ctUwfQ= -github.com/aws/aws-sdk-go-v2/service/lambda v1.89.0 h1:e4NAllPs/ygQ7W4dTlAuP5N7QpCT+rTij3S8UOv2DD4= -github.com/aws/aws-sdk-go-v2/service/lambda v1.89.0/go.mod h1:6HBXRyFFqOw+ALkJ6YGHfrr20/YXYv6X9pcZErXRvCA= github.com/aws/aws-sdk-go-v2/service/s3 v1.98.0 h1:foqo/ocQ7WqKwy3FojGtZQJo0FR4vto9qnz9VaumbCo= github.com/aws/aws-sdk-go-v2/service/s3 v1.98.0/go.mod h1:uoA43SdFwacedBfSgfFSjjCvYe8aYBS7EnU5GZ/YKMM= github.com/aws/aws-sdk-go-v2/service/signin v1.0.9 h1:QKZH0S178gCmFEgst8hN0mCX1KxLgHBKKY/CLqwP8lg= diff --git a/go.mod b/go.mod index f6a979e..5fd365a 100644 --- a/go.mod +++ b/go.mod @@ -7,7 +7,6 @@ require ( github.com/aws/aws-sdk-go-v2 v1.41.5 github.com/aws/aws-sdk-go-v2/config v1.32.14 github.com/aws/aws-sdk-go-v2/credentials v1.19.14 - github.com/aws/aws-sdk-go-v2/service/lambda v1.89.0 github.com/aws/aws-sdk-go-v2/service/s3 v1.98.0 github.com/aws/aws-sdk-go-v2/service/sqs v1.42.25 github.com/go-chi/chi/v5 v5.2.5 diff --git a/go.sum b/go.sum index c7a66b7..5177643 100644 --- a/go.sum +++ b/go.sum @@ -28,8 +28,6 @@ github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21 h1:c31//R3x github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.13.21/go.mod h1:r6+pf23ouCB718FUxaqzZdbpYFyDtehyZcmP5KL9FkA= github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21 h1:ZlvrNcHSFFWURB8avufQq9gFsheUgjVD9536obIknfM= github.com/aws/aws-sdk-go-v2/service/internal/s3shared v1.19.21/go.mod h1:cv3TNhVrssKR0O/xxLJVRfd2oazSnZnkUeTf6ctUwfQ= -github.com/aws/aws-sdk-go-v2/service/lambda v1.89.0 h1:e4NAllPs/ygQ7W4dTlAuP5N7QpCT+rTij3S8UOv2DD4= -github.com/aws/aws-sdk-go-v2/service/lambda v1.89.0/go.mod h1:6HBXRyFFqOw+ALkJ6YGHfrr20/YXYv6X9pcZErXRvCA= github.com/aws/aws-sdk-go-v2/service/s3 v1.98.0 h1:foqo/ocQ7WqKwy3FojGtZQJo0FR4vto9qnz9VaumbCo= github.com/aws/aws-sdk-go-v2/service/s3 v1.98.0/go.mod h1:uoA43SdFwacedBfSgfFSjjCvYe8aYBS7EnU5GZ/YKMM= github.com/aws/aws-sdk-go-v2/service/signin v1.0.9 h1:QKZH0S178gCmFEgst8hN0mCX1KxLgHBKKY/CLqwP8lg= diff --git a/internal/core/meter_test.go b/internal/core/meter_test.go index 629aad3..ca9dc2f 100644 --- a/internal/core/meter_test.go +++ b/internal/core/meter_test.go @@ -3,12 +3,30 @@ package core import ( "context" "encoding/json" + "net/http" + "net/http/httptest" "testing" + "github.com/mirrorstack-ai/app-module-sdk/auth" "github.com/mirrorstack-ai/app-module-sdk/meter" "github.com/mirrorstack-ai/app-module-sdk/system" ) +// usageStub starts a dispatch usage-ingress stand-in (202), points +// MS_DISPATCH_URL at it, and returns an app-scoped context. Record POSTs the +// usage Event here (the transport is dispatch-HTTP in dev + prod, mirroring +// ms.Emit), so a successful Record needs both an app-scoped ctx and a reachable +// dispatch. +func usageStub(t *testing.T) context.Context { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusAccepted) + })) + t.Cleanup(srv.Close) + t.Setenv("MS_DISPATCH_URL", srv.URL) + return auth.Set(context.Background(), auth.Identity{AppID: "a-456"}) +} + // TestMeter_DeclaresKindAndPriceIntoManifest asserts that ms.Meter records the // declared name + kind + unit + price into the manifest's metrics[] (the path // the platform reads to populate its metric_definitions catalog). Mirrors the @@ -117,10 +135,11 @@ func TestMeter_ReservedPriceOverrideInManifest(t *testing.T) { // TestRecord_RejectsReservedName asserts a reserved metric — even when declared // as a price-override — can never be self-reported via ms.Record. func TestRecord_RejectsReservedName(t *testing.T) { + ctx := usageStub(t) m, _ := New(Config{ID: "media"}) m.Meter("infra.compute.ms", meter.Price(0)) - if err := m.Record(context.Background(), "infra.compute.ms", 1); err == nil { + if err := m.Record(ctx, "infra.compute.ms", 1); err == nil { t.Error("expected an error recording a reserved platform-measured metric") } } @@ -136,10 +155,11 @@ func TestMeter_TopLevelPanicsBeforeInit(t *testing.T) { // for a DECLARED metric succeeds (dev client logs, returns nil), mirroring // ms.Emit's emit-by-name shape. func TestRecord_ResolvesDeclaredByName(t *testing.T) { + ctx := usageStub(t) m, _ := New(Config{ID: "media"}) m.Meter("orders.placed", meter.Counter, meter.Unit("order")) - if err := m.Record(context.Background(), "orders.placed", 1); err != nil { + if err := m.Record(ctx, "orders.placed", 1); err != nil { t.Fatalf("Record of a declared metric: %v", err) } } @@ -147,10 +167,11 @@ func TestRecord_ResolvesDeclaredByName(t *testing.T) { // TestRecord_RejectsUndeclaredName asserts declaration-first: ms.Record for a // name never declared via ms.Meter returns an error (no silent emit). func TestRecord_RejectsUndeclaredName(t *testing.T) { + ctx := usageStub(t) m, _ := New(Config{ID: "media"}) m.Meter("orders.placed", meter.Counter) - if err := m.Record(context.Background(), "never.declared", 1); err == nil { + if err := m.Record(ctx, "never.declared", 1); err == nil { t.Error("expected an error recording an undeclared metric name") } } diff --git a/internal/core/module.go b/internal/core/module.go index 30a1a72..eb3dcd3 100644 --- a/internal/core/module.go +++ b/internal/core/module.go @@ -110,7 +110,7 @@ type Module struct { // once" contract as internalAuth/proxyGuard — so every Platform() // registration reuses one closure instead of re-reading env + re-allocating // the closure (and its secret snapshot) on each call. - platformAuth func(http.Handler) http.Handler + platformAuth func(http.Handler) http.Handler poolCache *db.PoolCache // production: per-app DB pools devDBOnce sync.Once // dev mode: lazy DB init devDB *db.DB @@ -125,7 +125,7 @@ type Module struct { taskHandlers map[string]taskEntry // registered task handlers (startup-only writes) sqsClient *msqs.Client // nil in dev mode (MS_TASK_QUEUE_URL unset) signingKey []byte // HMAC key for TaskMessage signing (MS_TASK_SIGNING_KEY) - meterClient *meter.Client // prod (MS_METER_LAMBDA_ARN set) or dev-mode stderr + meterClient *meter.Client // dispatch-HTTP usage transport (dev + prod); never nil // devMode is true under `mirrorstack dev` (HTTP serving + dev DB), false in // Lambda/task-worker. It gates the dev-only per-app schema machinery: @@ -266,18 +266,15 @@ func New(cfg Config) (*Module, error) { m.sqsClient = sqsClient } - // Meter client: production mode when MS_METER_LAMBDA_ARN is set (ARN is - // validated at construction so typos fail fast), dev-mode stderr sink - // otherwise. Never nil. - if meterARN := os.Getenv("MS_METER_LAMBDA_ARN"); meterARN != "" { - meterClient, err := meter.NewFromARN(context.Background(), meterARN) - if err != nil { - return nil, fmt.Errorf("mirrorstack: init meter client: %w", err) - } - m.meterClient = meterClient - } else { - m.meterClient = meter.NewDev(m.logger) + // Meter client: dispatch-HTTP transport in both dev and prod (Record POSTs + // each usage Event to the dispatch usage ingress, like ms.Emit). The HTTP + // client is always built (never nil); New fail-fast validates MS_DISPATCH_URL + // when set so a typo surfaces at startup rather than as silently lost usage. + meterClient, err := meter.New() + if err != nil { + return nil, fmt.Errorf("mirrorstack: init meter client: %w", err) } + m.meterClient = meterClient // A Config-provided description flows to the registry so it reaches the // manifest like Name/Tags. Skip when empty to avoid a blank override. diff --git a/meter/meter.go b/meter/meter.go index ab33aa4..51c88fa 100644 --- a/meter/meter.go +++ b/meter/meter.go @@ -18,28 +18,36 @@ package meter import ( + "bytes" "context" "encoding/json" "fmt" - "log" + "io" "math" - "regexp" + "net/http" + "net/url" + "os" "strings" "sync" "time" - "github.com/aws/aws-sdk-go-v2/config" - "github.com/aws/aws-sdk-go-v2/service/lambda" - "github.com/aws/aws-sdk-go-v2/service/lambda/types" - "github.com/mirrorstack-ai/app-module-sdk/auth" "github.com/mirrorstack-ai/app-module-sdk/internal/ids" ) -// arnPattern matches a valid Lambda function ARN. Validated at Client -// construction so a typo in MS_METER_LAMBDA_ARN fails fast at startup -// rather than at first Record call (silent revenue loss otherwise). -var arnPattern = regexp.MustCompile(`^arn:aws:lambda:[a-z0-9-]+:[0-9]+:function:[a-zA-Z0-9_-]+$`) +// devDispatchFallback is the platform-dispatch base used when MS_DISPATCH_URL is +// unset. Modules run inside docker; the dispatch listens on the host at :8083, +// reachable via host.docker.internal — the same convention ms.Call / ms.Emit +// use (core.devDispatchFallback). The CLI sets MS_DISPATCH_URL explicitly in +// dev and prod points it at the real dispatch; this default only keeps local +// runs working. Restated here (not imported) because meter is a public package +// and must not import internal/core. +const devDispatchFallback = "http://host.docker.internal:8083" + +// meterTimeout bounds a single usage-POST to the dispatch. Matches the +// inter-module call timeout (core.callTimeout); metering is best-effort and must +// never hang a handler. +const meterTimeout = 15 * time.Second // reservedPrefixes are the platform-owned metric namespaces. A module may not // declare or self-report a metric under these — they belong to platform-side @@ -318,7 +326,12 @@ func (c *Client) Declare(moduleID string, decl Decl) { // Declaration-first: if name was never declared via ms.Meter, Record returns an // error (fail fast in dev) — it never silently emits an unknown metric. // -// Emitted via async Lambda invoke (production) or stderr log (dev mode). +// Emitted by POSTing the Event envelope to the platform dispatch usage ingress +// ({MS_DISPATCH_URL | dev fallback}/apps/{appID}/usage), exactly mirroring +// ms.Emit. Dispatch re-derives the authoritative app/module/owner/recorded_at +// and forwards to billing-engine; the SDK's Hint fields are untrusted. An empty +// app id (no auth identity in ctx) is an error (no panic), like ms.Emit. +// // Call sparingly — once per meaningful action, not per row processed. Errors // are returned, NOT panicked, and should be logged, not propagated: a billing // failure must never fail the handler. @@ -327,9 +340,9 @@ func (c *Client) Declare(moduleID string, decl Decl) { // a single bad value can't crash the handler. The metric name was already // validated at declaration. // -// The EventID is minted ONCE per Record call and reused across any transport -// retry within the call, so the platform's ON CONFLICT(event_id) dedupe holds -// for a retried delivery (the retry loop lands with the transport rewrite). +// The EventID is minted ONCE per Record call (and the built Event reused across +// any transport retry of the same call) so the platform's ON CONFLICT(event_id) +// dedupe holds and a retried delivery is not double-counted. func (c *Client) Record(ctx context.Context, name string, value float64) error { c.mu.RLock() decl, declared := c.metrics[name] @@ -353,20 +366,15 @@ func (c *Client) Record(ctx context.Context, name string, value float64) error { if a := auth.Get(ctx); a != nil { appID = a.AppID } - - // Dev mode: log to stderr and return. appID may be empty if the context - // has no auth identity (internal route, test harness). A zero Client{} (or - // one built via NewFromARN before the logger is wired) may have a nil - // logger, so guard it — a missing dev sink must not panic the handler. - if c.lambdaClient == nil { - if c.logger != nil { - c.logger.Printf("meter: appID=%q moduleID=%q metric=%q value=%g", appID, moduleID, decl.Name, value) - } - return nil + // An empty app id means there is no app scope to attribute the usage to. + // Mirror ms.Emit: return an error (no panic), and do not reach the transport. + if appID == "" { + return fmt.Errorf("mirrorstack/meter: Record requires an app-scoped context (no AppID in auth identity)") } - // Mint the EventID ONCE here so a retried delivery reuses it. NO kind on - // the wire — the manifest/catalog is authoritative. + // Mint the EventID ONCE here so a retried delivery reuses it and the + // platform's ON CONFLICT(event_id) dedupe holds. NO kind on the wire — the + // manifest/catalog is authoritative. event := Event{ V: envelopeVersion, EventID: ids.NewUUID(), @@ -376,71 +384,98 @@ func (c *Client) Record(ctx context.Context, name string, value float64) error { Value: value, RecordedAtHint: time.Now().UTC(), } - return c.dispatch(ctx, event) + return c.dispatch(ctx, appID, event) } -// dispatch delivers an already-built Event to the transport. The caller mints -// the EventID once (in Record), so retrying dispatch with the same Event reuses -// the same EventID and the platform deduplicates rather than double-counts. -func (c *Client) dispatch(ctx context.Context, event Event) error { +// resolveUsageURL builds the platform-dispatch usage-ingress URL the Event is +// POSTed to: +// +// {base}/apps/{appID}/usage +// +// base is MS_DISPATCH_URL (the container->dispatch base) with the +// host.docker.internal:8083 dev fallback when unset — the same resolution +// ms.Call / ms.Emit use. Dispatch re-derives the authoritative app/module from +// the authenticated invoker; the appID in the path is the SDK's hint scope. +// +// DEV/DISPATCH TRANSPORT. The prod module->dispatch leg rides task #146 (the +// same seam ms.Emit's resolveEventBusURL documents); the dispatch->billing leg +// is already prod-ready. When #146 lands, swap the body of this function (and +// only this function) to consult the prod transport; Record's +// marshal/auth/error contract stays put. +func resolveUsageURL(appID string) string { + base := os.Getenv("MS_DISPATCH_URL") + if base == "" { + base = devDispatchFallback + } + base = strings.TrimRight(base, "/") + return fmt.Sprintf("%s/apps/%s/usage", base, appID) +} + +// dispatch delivers an already-built Event to the platform dispatch usage +// ingress over HTTP, mirroring ms.Emit. The caller (Record) mints the EventID +// once, so retrying dispatch with the same Event reuses the same EventID and the +// platform deduplicates rather than double-counts. A non-2xx response is an +// error with the body truncated to ~2 KB; the non-fatal contract (log, don't +// propagate) is the caller's responsibility. +func (c *Client) dispatch(ctx context.Context, appID string, event Event) error { body, err := json.Marshal(event) if err != nil { return fmt.Errorf("mirrorstack/meter: marshal event: %w", err) } - _, err = c.lambdaClient.Invoke(ctx, &lambda.InvokeInput{ - FunctionName: &c.functionARN, - InvocationType: types.InvocationTypeEvent, // async fire-and-forget - Payload: body, - }) + + u := resolveUsageURL(appID) + req, err := http.NewRequestWithContext(ctx, http.MethodPost, u, bytes.NewReader(body)) if err != nil { - return fmt.Errorf("mirrorstack/meter: invoke %s: %w", c.functionARN, err) + return err } - return nil -} + req.Header.Set("Content-Type", "application/json") + req.Header.Set("X-MS-App-ID", appID) -// lambdaInvoker is the subset of lambda.Client used by Client. Makes the -// Lambda invoke path mockable in unit tests. -type lambdaInvoker interface { - Invoke(ctx context.Context, params *lambda.InvokeInput, optFns ...func(*lambda.Options)) (*lambda.InvokeOutput, error) + resp, err := c.httpClient.Do(req) + if err != nil { + return err + } + defer resp.Body.Close() + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + b, _ := io.ReadAll(io.LimitReader(resp.Body, 2048)) + return fmt.Errorf("ms.Record %s -> %d: %s", req.URL.Path, resp.StatusCode, strings.TrimSpace(string(b))) + } + return nil } // Client is the module-level meter transport AND metric registry. Created -// eagerly at ms.Init: a production client (MS_METER_LAMBDA_ARN set) -// async-invokes the platform meter Lambda; a dev client logs to stderr. -// Declared metrics live in its registry, keyed by name, so Record resolves a -// metric by name (mirroring ms.Emits/ms.Emit). +// eagerly at ms.Init: it POSTs each usage Event to the platform dispatch usage +// ingress over HTTP (mirroring ms.Emit / ms.Call), in both dev and prod — there +// is no separate dev sink. Declared metrics live in its registry, keyed by name, +// so Record resolves a metric by name (mirroring ms.Emits/ms.Emit). // -// The dispatch-HTTP transport rewrite is a follow-up (PR #2); this PR keeps -// the existing MS_METER_LAMBDA_ARN / lambda.Invoke transport. +// The transport is dispatch-HTTP: MS_DISPATCH_URL (or the dev fallback) routes +// to the dispatch usage ingress, which re-derives the authoritative attribution +// and forwards to billing-engine. The HTTP client is ALWAYS built (never nil) — +// in dev it harmlessly targets the local dispatch, and a transport failure is +// non-fatal (returned, then logged by the caller, not propagated). type Client struct { - lambdaClient lambdaInvoker - functionARN string - logger *log.Logger // dev-mode stderr sink when lambdaClient is nil + httpClient *http.Client mu sync.RWMutex moduleID string // emitting module's Config.ID, set at first Declare metrics map[string]Decl // declared metrics, keyed by name (Record resolves here) } -// NewFromARN creates a production meter client for the given Lambda function -// ARN. Validates ARN format against arnPattern. Uses the module's default -// AWS IAM role (same pattern as internal/sqs/client.go). -func NewFromARN(ctx context.Context, arn string) (*Client, error) { - if !arnPattern.MatchString(arn) { - return nil, fmt.Errorf("mirrorstack/meter: invalid ARN format %q (expected arn:aws:lambda:::function:)", arn) - } - cfg, err := config.LoadDefaultConfig(ctx) - if err != nil { - return nil, fmt.Errorf("mirrorstack/meter: load aws config: %w", err) +// New creates a meter client. The transport is dispatch-HTTP in both dev and +// prod (the URL is resolved per-Record from MS_DISPATCH_URL with the dev +// fallback), so there is a single constructor — no ARN/dev split. The HTTP +// client is always built so Record never dereferences a nil transport. +// +// Fail-fast prod-base validation mirrors ms.Emit: if MS_DISPATCH_URL is set it +// must be a parseable http(s) URL, so a typo fails at startup rather than at the +// first (silently lost) Record call. An unset MS_DISPATCH_URL is valid (dev +// fallback). +func New() (*Client, error) { + if base := os.Getenv("MS_DISPATCH_URL"); base != "" { + if u, err := url.Parse(base); err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" { + return nil, fmt.Errorf("mirrorstack/meter: MS_DISPATCH_URL %q is not a valid http(s) base URL", base) + } } - return &Client{ - lambdaClient: lambda.NewFromConfig(cfg), - functionARN: arn, - }, nil -} - -// NewDev creates a dev-mode meter client that logs Record calls to the given -// logger (typically Module.logger writing to stderr). -func NewDev(logger *log.Logger) *Client { - return &Client{logger: logger} + return &Client{httpClient: &http.Client{Timeout: meterTimeout}}, nil } diff --git a/meter/meter_test.go b/meter/meter_test.go index 0d1ef86..3c537ab 100644 --- a/meter/meter_test.go +++ b/meter/meter_test.go @@ -1,42 +1,76 @@ package meter import ( - "bytes" "context" "encoding/json" - "errors" - "log" + "io" "math" + "net/http" + "net/http/httptest" "strings" + "sync" "testing" - "github.com/aws/aws-sdk-go-v2/service/lambda" - "github.com/aws/aws-sdk-go-v2/service/lambda/types" - "github.com/mirrorstack-ai/app-module-sdk/auth" ) -type fakeLambda struct { - invoked int - lastIn *lambda.InvokeInput - err error +// capture records what the dispatch usage ingress received for a Record POST. +type capture struct { + mu sync.Mutex + hits int + method string + path string + appID string + ct string + body []byte } -func (f *fakeLambda) Invoke(ctx context.Context, in *lambda.InvokeInput, _ ...func(*lambda.Options)) (*lambda.InvokeOutput, error) { - f.invoked++ - f.lastIn = in - if f.err != nil { - return nil, f.err - } - return &lambda.InvokeOutput{StatusCode: 202}, nil +func (c *capture) get() capture { + c.mu.Lock() + defer c.mu.Unlock() + return capture{hits: c.hits, method: c.method, path: c.path, appID: c.appID, ct: c.ct, body: c.body} } -func newTestClient(t *testing.T, fake *fakeLambda) *Client { +// newDispatchStub starts an httptest server standing in for the dispatch usage +// ingress, points MS_DISPATCH_URL at it, and returns a client + the capture. The +// status code the stub returns is configurable for the non-2xx test. +func newDispatchStub(t *testing.T, status int) (*Client, *capture) { t.Helper() - return &Client{ - lambdaClient: fake, - functionARN: "arn:aws:lambda:us-east-1:123456789012:function:meter-test", + cap := &capture{} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cap.mu.Lock() + cap.hits++ + cap.method = r.Method + cap.path = r.URL.Path + cap.appID = r.Header.Get("X-MS-App-ID") + cap.ct = r.Header.Get("Content-Type") + cap.body, _ = io.ReadAll(r.Body) + cap.mu.Unlock() + if status == 0 { + status = http.StatusAccepted + } + w.WriteHeader(status) + if status >= 300 { + _, _ = w.Write([]byte("usage ingress unavailable")) + } + })) + t.Cleanup(srv.Close) + t.Setenv("MS_DISPATCH_URL", srv.URL) + c, err := New() + if err != nil { + t.Fatalf("New: %v", err) } + return c, cap +} + +// newTestClient returns a meter client with a real HTTP transport but no +// configured dispatch (it targets whatever MS_DISPATCH_URL / dev fallback +// resolves to). Used by tests that assert on the pre-transport path (validation, +// declaration) and must NOT reach a server — those tests use an undeclared / +// reserved / invalid input so dispatch is never called. +func newTestClient(t *testing.T) *Client { + t.Helper() + return &Client{httpClient: &http.Client{}} } // declareCounter is a test helper: declares a counter metric on c bound to @@ -46,58 +80,84 @@ func declareCounter(t *testing.T, c *Client, name string) { c.Declare("media", DeclFromOptions(name, Counter)) } -func TestNewFromARN_ValidARN(t *testing.T) { - // config.LoadDefaultConfig probes IMDS for region/credentials; disable it so - // the test is hermetic and skips the ~100ms IMDS round-trip in CI/sandboxes - // without AWS credentials. (Not t.Parallel: t.Setenv forbids parallel.) - t.Setenv("AWS_EC2_METADATA_DISABLED", "true") - _, err := NewFromARN(context.Background(), "arn:aws:lambda:us-east-1:123456789012:function:meter") +// appCtx is a context carrying an auth identity with the given app id, the +// scope Record needs to attribute usage. +func appCtx(appID string) context.Context { + return auth.Set(context.Background(), auth.Identity{AppID: appID}) +} + +func TestNew_RejectsMalformedDispatchURL(t *testing.T) { + for _, bad := range []string{"://nope", "not a url", "ftp://host", "http://"} { + t.Run(bad, func(t *testing.T) { + t.Setenv("MS_DISPATCH_URL", bad) + if _, err := New(); err == nil { + t.Errorf("New() should reject MS_DISPATCH_URL=%q", bad) + } + }) + } +} + +func TestNew_AllowsUnsetDispatchURL(t *testing.T) { + t.Setenv("MS_DISPATCH_URL", "") + c, err := New() if err != nil { - t.Fatalf("NewFromARN with valid ARN: %v", err) + t.Fatalf("New() with unset MS_DISPATCH_URL should succeed (dev fallback): %v", err) + } + if c.httpClient == nil { + t.Error("New() must always build the HTTP client (never nil)") } } -func TestNewFromARN_InvalidARN(t *testing.T) { - t.Parallel() - cases := []string{ - "", - "not-an-arn", - "arn:aws:s3:::bucket/key", // wrong service - "arn:aws:lambda::123456789012:function:meter", // missing region - "arn:aws:lambda:us-east-1::function:meter", // missing account - "arn:aws:lambda:us-east-1:123456789012:function:", // missing name - "arn:aws:lambda:us-east-1:123456789012:function:bad name", // space in name - } - for _, arn := range cases { - t.Run(arn, func(t *testing.T) { - _, err := NewFromARN(context.Background(), arn) - if err == nil { - t.Errorf("NewFromARN(%q) should reject invalid ARN", arn) +func TestResolveUsageURL_Building(t *testing.T) { + cases := []struct { + name string + dispatch string // "" = unset -> dev fallback + appID string + want string + }{ + {"dev fallback when unset", "", "a-456", devDispatchFallback + "/apps/a-456/usage"}, + {"explicit base", "http://dispatch:8083", "a-456", "http://dispatch:8083/apps/a-456/usage"}, + {"trailing slash trimmed", "http://dispatch:8083/", "a-456", "http://dispatch:8083/apps/a-456/usage"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Setenv("MS_DISPATCH_URL", tc.dispatch) + if got := resolveUsageURL(tc.appID); got != tc.want { + t.Errorf("resolveUsageURL(%q) = %q, want %q", tc.appID, got, tc.want) } }) } } -func TestRecord_ProdInvokesLambda(t *testing.T) { - t.Parallel() - fake := &fakeLambda{} - c := newTestClient(t, fake) +func TestRecord_PostsEventToUsageIngress(t *testing.T) { + c, cap := newDispatchStub(t, http.StatusAccepted) declareCounter(t, c, "transcode.minutes") ctx := auth.Set(context.Background(), auth.Identity{AppID: "app_abc", AppRole: "admin"}) if err := c.Record(ctx, "transcode.minutes", 12); err != nil { t.Fatalf("Record: %v", err) } - if fake.invoked != 1 { - t.Errorf("Lambda invoked %d times, want 1", fake.invoked) + + g := cap.get() + if g.hits != 1 { + t.Errorf("usage ingress hit %d times, want 1", g.hits) + } + if g.method != http.MethodPost { + t.Errorf("method = %q, want POST", g.method) + } + if g.path != "/apps/app_abc/usage" { + t.Errorf("path = %q, want /apps/app_abc/usage", g.path) } - if fake.lastIn.InvocationType != types.InvocationTypeEvent { - t.Errorf("invocation type = %v, want Event (async)", fake.lastIn.InvocationType) + if g.appID != "app_abc" { + t.Errorf("X-MS-App-ID = %q, want app_abc", g.appID) + } + if g.ct != "application/json" { + t.Errorf("Content-Type = %q, want application/json", g.ct) } var got Event - if err := json.Unmarshal(fake.lastIn.Payload, &got); err != nil { - t.Fatalf("decode payload: %v", err) + if err := json.Unmarshal(g.body, &got); err != nil { + t.Fatalf("decode payload: %v (body=%s)", err, g.body) } if got.V != 1 { t.Errorf("envelope version = %d, want 1", got.V) @@ -122,6 +182,39 @@ func TestRecord_ProdInvokesLambda(t *testing.T) { } } +func TestRecord_EmptyAppContextErrorsWithoutHTTP(t *testing.T) { + c, cap := newDispatchStub(t, http.StatusAccepted) + declareCounter(t, c, "transcode.minutes") + + // No auth identity on the context -> no AppID -> error, no HTTP call. + if err := c.Record(context.Background(), "transcode.minutes", 1); err == nil { + t.Fatal("expected error on empty-app context, got nil") + } + if cap.get().hits != 0 { + t.Error("Record made an HTTP call despite empty app context") + } +} + +func TestRecord_Non2xxReturnsErrorWithTruncatedBody(t *testing.T) { + c, _ := newDispatchStub(t, http.StatusBadGateway) + declareCounter(t, c, "transcode.minutes") + + err := c.Record(appCtx("a-456"), "transcode.minutes", 1) + if err == nil { + t.Fatal("expected error on non-2xx, got nil") + } + msg := err.Error() + if !strings.Contains(msg, "502") { + t.Errorf("error %q missing status 502", msg) + } + if !strings.Contains(msg, "usage ingress unavailable") { + t.Errorf("error %q missing upstream body", msg) + } + if !strings.Contains(msg, "/apps/a-456/usage") { + t.Errorf("error %q missing request path", msg) + } +} + // TestMeter_DeclaresKindAndPriceIntoManifest asserts the declaration carries the // kind / unit / price the platform populates metric_definitions from (the core // ms.Meter facade hands this Decl to both registry.AddMetric and the meter @@ -158,7 +251,7 @@ func TestMeter_DeclaresKindAndPriceIntoManifest(t *testing.T) { // of PR #1b: kind moved from a positional argument to a functional option. func TestMeter_KindIsAnOption(t *testing.T) { t.Parallel() - c := newTestClient(t, &fakeLambda{}) + c := newTestClient(t) // Order-independent: kind option can sit anywhere in the variadic list. c.Declare("media", DeclFromOptions("orders.placed", Unit("order"), Counter, Price(50_000))) c.mu.RLock() @@ -173,7 +266,7 @@ func TestMeter_KindIsAnOption(t *testing.T) { // option panics — the platform must know SUM vs MAX/integral up front. func TestMeter_RejectsMissingKind(t *testing.T) { t.Parallel() - c := newTestClient(t, &fakeLambda{}) + c := newTestClient(t) defer func() { if r := recover(); r == nil { t.Error("expected panic on a custom metric declared without a kind") @@ -186,7 +279,7 @@ func TestMeter_RejectsMissingKind(t *testing.T) { // panics — a metric has exactly one kind. func TestMeter_RejectsConflictingKinds(t *testing.T) { t.Parallel() - c := newTestClient(t, &fakeLambda{}) + c := newTestClient(t) defer func() { if r := recover(); r == nil { t.Error("expected panic when both Counter and Gauge are passed") @@ -200,7 +293,7 @@ func TestMeter_RejectsConflictingKinds(t *testing.T) { // and lands in the registry with NO kind/unit (platform-owned), price set. func TestMeter_ReservedPriceOnlyAccepted(t *testing.T) { t.Parallel() - c := newTestClient(t, &fakeLambda{}) + c := newTestClient(t) c.Declare("media", DeclFromOptions("infra.compute.ms", Price(0))) c.mu.RLock() got, ok := c.metrics["infra.compute.ms"] @@ -230,7 +323,7 @@ func TestMeter_ReservedWithKindOrUnitPanics(t *testing.T) { } for label, opts := range cases { t.Run(label, func(t *testing.T) { - c := newTestClient(t, &fakeLambda{}) + c := newTestClient(t) defer func() { if r := recover(); r == nil { t.Errorf("expected panic on reserved metric with a %s option", label) @@ -247,7 +340,7 @@ func TestMeter_ReservedWithKindOrUnitPanics(t *testing.T) { // meaningless no-op that would otherwise pollute the manifest. func TestMeter_ReservedWithoutPricePanics(t *testing.T) { t.Parallel() - c := newTestClient(t, &fakeLambda{}) + c := newTestClient(t) defer func() { if r := recover(); r == nil { t.Error("expected panic on a reserved metric declared with no ms.Price") @@ -260,42 +353,39 @@ func TestMeter_ReservedWithoutPricePanics(t *testing.T) { // price-override but can never self-report its value — ms.Record returns an // error and never reaches the transport (the platform meters infra itself). func TestRecord_RejectsReservedName(t *testing.T) { - t.Parallel() - fake := &fakeLambda{} - c := newTestClient(t, fake) + c, cap := newDispatchStub(t, http.StatusAccepted) c.Declare("media", DeclFromOptions("infra.compute.ms", Price(0))) - err := c.Record(context.Background(), "infra.compute.ms", 1) + err := c.Record(appCtx("a-456"), "infra.compute.ms", 1) if err == nil || !strings.Contains(err.Error(), "platform-measured") { t.Errorf("expected a platform-measured rejection, got %v", err) } - if fake.invoked != 0 { - t.Errorf("reserved metric must not reach the transport; invoked=%d", fake.invoked) + if cap.get().hits != 0 { + t.Errorf("reserved metric must not reach the transport; hits=%d", cap.get().hits) } } // TestRecord_NoKindOnWire asserts the §4 invariant: kind lives in the manifest, -// NOT on the wire. The serialized Event must carry no "kind" key (for either a -// counter or a gauge), the envelope version must stay 1, and the value is -// carried verbatim (V==1). +// NOT on the wire. The serialized Event POSTed to the usage ingress must carry +// no "kind" key (for either a counter or a gauge), the envelope version must +// stay 1, and the value is carried verbatim. func TestRecord_NoKindOnWire(t *testing.T) { - t.Parallel() for _, kindOpt := range []MetricOption{Counter, Gauge} { - fake := &fakeLambda{} - c := newTestClient(t, fake) + c, cap := newDispatchStub(t, http.StatusAccepted) c.Declare("store", DeclFromOptions("myapp.items", kindOpt)) - if err := c.Record(context.Background(), "myapp.items", 1); err != nil { + if err := c.Record(appCtx("a-456"), "myapp.items", 1); err != nil { t.Fatalf("Record: %v", err) } + body := cap.get().body var raw map[string]json.RawMessage - if err := json.Unmarshal(fake.lastIn.Payload, &raw); err != nil { + if err := json.Unmarshal(body, &raw); err != nil { t.Fatalf("decode payload: %v", err) } if _, ok := raw["kind"]; ok { t.Errorf("wire envelope must not carry a kind field, got keys %v", keys(raw)) } var got Event - if err := json.Unmarshal(fake.lastIn.Payload, &got); err != nil { + if err := json.Unmarshal(body, &got); err != nil { t.Fatalf("decode event: %v", err) } if got.V != 1 { @@ -315,98 +405,34 @@ func keys(m map[string]json.RawMessage) []string { return out } -func TestRecord_PropagatesLambdaError(t *testing.T) { - t.Parallel() - fake := &fakeLambda{err: errors.New("throttled")} - c := newTestClient(t, fake) - declareCounter(t, c, "transcode.minutes") - - err := c.Record(context.Background(), "transcode.minutes", 1) - if err == nil || !strings.Contains(err.Error(), "throttled") { - t.Errorf("expected wrapped throttled error, got %v", err) - } -} - // TestRecord_RejectsUndeclaredName asserts the declaration-first contract: a // Record for a name never declared via ms.Meter returns an error and never // reaches the transport. func TestRecord_RejectsUndeclaredName(t *testing.T) { - t.Parallel() - fake := &fakeLambda{} - c := newTestClient(t, fake) + c, cap := newDispatchStub(t, http.StatusAccepted) declareCounter(t, c, "transcode.minutes") - err := c.Record(context.Background(), "never.declared", 1) + err := c.Record(appCtx("a-456"), "never.declared", 1) if err == nil || !strings.Contains(err.Error(), "never declared") { t.Errorf("expected an undeclared-name error, got %v", err) } - if fake.invoked != 0 { - t.Errorf("undeclared metric must not reach the transport; invoked=%d", fake.invoked) - } -} - -func TestRecord_DevModeLogsToStderr(t *testing.T) { - var buf bytes.Buffer - c := NewDev(log.New(&buf, "", 0)) - c.Declare("media", DeclFromOptions("transcode.minutes", Counter)) - - ctx := auth.Set(context.Background(), auth.Identity{AppID: "dev_app"}) - if err := c.Record(ctx, "transcode.minutes", 12); err != nil { - t.Fatalf("Record: %v", err) - } - - out := buf.String() - if !strings.Contains(out, `appID="dev_app"`) || - !strings.Contains(out, `moduleID="media"`) || - !strings.Contains(out, `metric="transcode.minutes"`) || - !strings.Contains(out, `value=12`) { - t.Errorf("unexpected log line: %q", out) - } - if strings.Contains(out, "kind=") { - t.Errorf("dev log must not carry a kind (catalog is authoritative): %q", out) - } -} - -func TestRecord_NilLoggerDoesNotPanic(t *testing.T) { - t.Parallel() - // A zero Client{} (legal via the exported type, reachable in tests/mocks) - // has both lambdaClient and logger nil. Record must take the dev-mode path - // (lambdaClient == nil) and return without dereferencing the nil logger. - c := &Client{} - c.Declare("media", DeclFromOptions("transcode.minutes", Counter)) - - ctx := auth.Set(context.Background(), auth.Identity{AppID: "dev_app"}) - if err := c.Record(ctx, "transcode.minutes", 7); err != nil { - t.Fatalf("Record with nil logger should be a no-op, got: %v", err) - } -} - -func TestRecord_DevMode_EmptyAppIDWhenNoAuth(t *testing.T) { - var buf bytes.Buffer - c := NewDev(log.New(&buf, "", 0)) - c.Declare("media", DeclFromOptions("transcode.minutes", Counter)) - - _ = c.Record(context.Background(), "transcode.minutes", 1) - - if !strings.Contains(buf.String(), `appID=""`) { - t.Errorf("expected appID=\"\" when context has no auth identity, got: %q", buf.String()) + if cap.get().hits != 0 { + t.Errorf("undeclared metric must not reach the transport; hits=%d", cap.get().hits) } } func TestRecord_RejectsNegativeAndNonFinite(t *testing.T) { - t.Parallel() - fake := &fakeLambda{} - c := newTestClient(t, fake) + c, cap := newDispatchStub(t, http.StatusAccepted) declareCounter(t, c, "transcode.minutes") bad := []float64{-1, math.NaN(), math.Inf(1), math.Inf(-1)} for _, v := range bad { - if err := c.Record(context.Background(), "transcode.minutes", v); err == nil { + if err := c.Record(appCtx("a-456"), "transcode.minutes", v); err == nil { t.Errorf("Record(%g) should return an error (finite, non-negative)", v) } } - if fake.invoked != 0 { - t.Errorf("invalid values must not reach the transport; invoked=%d", fake.invoked) + if cap.get().hits != 0 { + t.Errorf("invalid values must not reach the transport; hits=%d", cap.get().hits) } } @@ -419,7 +445,7 @@ func TestMeter_ReservedKindOnAnyPrefixPanics(t *testing.T) { bad := []string{"infra.compute.ms", "infra.egress.bytes", "platform.storage.bytes", "platform.tokens"} for _, name := range bad { t.Run(name, func(t *testing.T) { - c := newTestClient(t, &fakeLambda{}) + c := newTestClient(t) defer func() { if r := recover(); r == nil { t.Errorf("expected panic on a kind option for reserved metric %q", name) @@ -435,7 +461,7 @@ func TestMeter_RejectsInvalidName(t *testing.T) { bad := []string{"", "has/slash", "has space", "has..dots", "null\x00byte"} for _, name := range bad { t.Run(name, func(t *testing.T) { - c := newTestClient(t, &fakeLambda{}) + c := newTestClient(t) defer func() { if r := recover(); r == nil { t.Errorf("expected panic on invalid metric name %q", name) @@ -448,7 +474,7 @@ func TestMeter_RejectsInvalidName(t *testing.T) { func TestMeter_RejectsInvalidKind(t *testing.T) { t.Parallel() - c := newTestClient(t, &fakeLambda{}) + c := newTestClient(t) defer func() { if r := recover(); r == nil { t.Error("expected panic on invalid kind") @@ -464,7 +490,7 @@ func TestMeter_RejectsInvalidKind(t *testing.T) { // second declaration would silently disagree on kind/price. func TestMeter_RejectsDuplicateName(t *testing.T) { t.Parallel() - c := newTestClient(t, &fakeLambda{}) + c := newTestClient(t) c.Declare("media", DeclFromOptions("orders.placed", Counter)) defer func() { if r := recover(); r == nil { @@ -480,29 +506,28 @@ func TestMeter_RejectsDuplicateName(t *testing.T) { // double-counted. We Record once (mints the EventID), then re-dispatch the same // built event (simulating a transport retry) and assert the EventID is stable. func TestEventID_StableAcrossRetry(t *testing.T) { - t.Parallel() - fake := &fakeLambda{} - c := newTestClient(t, fake) + c, cap := newDispatchStub(t, http.StatusAccepted) declareCounter(t, c, "transcode.minutes") - if err := c.Record(context.Background(), "transcode.minutes", 1); err != nil { + // Record once: mints the EventID and POSTs the first attempt. + if err := c.Record(appCtx("a-456"), "transcode.minutes", 1); err != nil { t.Fatalf("Record: %v", err) } var first Event - if err := json.Unmarshal(fake.lastIn.Payload, &first); err != nil { + if err := json.Unmarshal(cap.get().body, &first); err != nil { t.Fatalf("decode first payload: %v", err) } if first.EventID == "" { t.Fatal("EventID should be set") } - // Retry of the SAME logical call must reuse the SAME EventID. dispatch is - // the per-attempt transport leg; Record owns the minted event. - if err := c.dispatch(context.Background(), first); err != nil { + // Retry of the SAME logical call must reuse the SAME EventID and app id. + // dispatch is the per-attempt transport leg; Record owns the minted event. + if err := c.dispatch(appCtx("a-456"), "a-456", first); err != nil { t.Fatalf("dispatch retry: %v", err) } var retried Event - if err := json.Unmarshal(fake.lastIn.Payload, &retried); err != nil { + if err := json.Unmarshal(cap.get().body, &retried); err != nil { t.Fatalf("decode retry payload: %v", err) } if retried.EventID != first.EventID {