From 3b80d54c4c2f14f22b40664b975d44d3967e8ee3 Mon Sep 17 00:00:00 2001 From: Sheng Kun Chang Date: Mon, 6 Jul 2026 11:38:25 +0800 Subject: [PATCH] feat: add platform dev-seed lifecycle endpoint POST /__mirrorstack/platform/lifecycle/app/seed (dev/tunnel-only, gated by internalAuth + !runtime.IsLambda) lets the platform push a targeted per-app-per-module data snapshot into a dev-mounted module's local schema over the authenticated tunnel: COPY-text ingestion via PgConn().CopyFrom, if-empty guard so existing local dev writes are preserved, optional CreateSQL to materialize a producer's exposure-anchored dependency tables, sanitized identifiers, 2MB body cap. Enables a deployed-installed module to be dev-mounted with its real data (F2) without a full-stack copy. Adds an 8MB group backstop on /platform so a future route can't be uncapped. Co-Authored-By: Claude Opus 4.8 (1M context) --- db/pool_cache.go | 9 ++ internal/core/db.go | 29 ++++ internal/core/module.go | 65 ++++++-- internal/core/module_test.go | 53 +++++++ internal/runtime/lambda.go | 18 +++ system/lifecycle.go | 153 ++++++++++++++++++ system/seed_integration_test.go | 271 ++++++++++++++++++++++++++++++++ system/seed_test.go | 137 ++++++++++++++++ 8 files changed, 718 insertions(+), 17 deletions(-) create mode 100644 system/seed_integration_test.go create mode 100644 system/seed_test.go diff --git a/db/pool_cache.go b/db/pool_cache.go index 80ab014..0d84ab3 100644 --- a/db/pool_cache.go +++ b/db/pool_cache.go @@ -129,6 +129,15 @@ func afterReleaseReset(conn *pgx.Conn) bool { // it uses session-scoped SET / set_config(_, _, false). The Tx() function uses // transaction-local SET LOCAL inside its BEGIN block. func AcquireScoped(ctx context.Context, pool *pgxpool.Pool) (Querier, func(), error) { + return AcquireScopedConn(ctx, pool) +} + +// AcquireScopedConn is AcquireScoped but returns the concrete *pgxpool.Conn +// instead of the narrower Querier interface. Callers that need the raw driver +// connection underneath — e.g. COPY FROM STDIN via conn.Conn().PgConn(), +// which db.Querier does not expose — use this; everything else should keep +// using AcquireScoped. +func AcquireScopedConn(ctx context.Context, pool *pgxpool.Pool) (*pgxpool.Conn, func(), error) { conn, err := pool.Acquire(ctx) if err != nil { return nil, nil, fmt.Errorf("mirrorstack/db: failed to acquire connection: %w", err) diff --git a/internal/core/db.go b/internal/core/db.go index 2bc779d..2248a32 100644 --- a/internal/core/db.go +++ b/internal/core/db.go @@ -65,6 +65,35 @@ func (m *Module) resolvePool(ctx context.Context) (*pgxpool.Pool, func(), error) return m.resolvePoolFor(ctx, db.CredentialFrom) } +// seedConn resolves an app-schema-scoped connection for system.SeedHandler +// (the dev-mount seed endpoint) WITHOUT the cross-module dev guard that +// Module.DB/Tx apply. The seed endpoint legitimately COPYs into another +// module's exposure-anchored dependency table — the platform already +// enforced that exposure-anchor grant before it ever sent the chunk (see +// devseed.Seeder.dependencyTables on the platform side) — so routing it +// through devGuardFor would reject every dependency-table seed in dev mode +// with a false-positive cross-module error. +// +// Also unlike Module.DB, this returns the concrete *pgxpool.Conn (via +// db.AcquireScopedConn) rather than the narrower db.Querier: COPY FROM STDIN +// needs the raw driver connection's PgConn(), which db.Querier does not +// expose. +func (m *Module) seedConn(ctx context.Context) (*pgxpool.Conn, func(), error) { + pool, releasePool, err := m.resolvePool(ctx) + if err != nil { + return nil, nil, err + } + conn, releaseConn, err := db.AcquireScopedConn(ctx, pool) + if err != nil { + releasePool() + return nil, nil, err + } + return conn, func() { + releaseConn() + releasePool() + }, nil +} + // resolvePoolFor is the shared implementation behind resolvePool and // resolveModulePool. Production reads a credential from the context via // getCred (different context key per scope) and pulls a refcount-pinned diff --git a/internal/core/module.go b/internal/core/module.go index 886d2a4..ac5de3b 100644 --- a/internal/core/module.go +++ b/internal/core/module.go @@ -148,18 +148,14 @@ type Module struct { devProvision sync.Map } -// devAppSchemaName derives the per-app Postgres schema from an app id, matching -// the platform's ids.AppSchemaName convention (app_). -// It validates against the same pattern the production Lambda shim uses -// (runtime.AppSchemaPattern) so dev and prod agree on the shape; -// pgx.Identifier.Sanitize is the second line of defense before SQL. Returns -// ok=false if the result is not a valid schema identifier. +// devAppSchemaName derives the per-app Postgres schema from an app id, +// delegating to runtime.AppSchemaName so this package and system.SeedHandler +// (which derives the same schema from the dev-mount seed contract's bare +// appId) can never drift on the shape. pgx.Identifier.Sanitize is the second +// line of defense before SQL. Returns ok=false if the result is not a valid +// schema identifier. func devAppSchemaName(appID string) (string, bool) { - schema := "app_" + strings.ReplaceAll(strings.ToLower(appID), "-", "_") - if !runtime.AppSchemaPattern.MatchString(schema) { - return "", false - } - return schema, true + return runtime.AppSchemaName(appID) } // devAppSchemaMiddleware is the dev analog of the Lambda shim's per-request @@ -793,9 +789,25 @@ func (m *Module) mountSystemRoutes() { }) r.Route("/platform", func(r chi.Router) { - r.Use(httputil.MaxBytes(64 * 1024)) // 64 KB — lifecycle bodies are tiny r.Use(m.internalAuth) - r.Get("/manifest", system.ManifestHandler( + // 8 MB group-wide backstop. Defense-in-depth only: every route + // below applies its own tighter cap (64 KB via smallBody, or + // seed's own 2 MB reader — see system/lifecycle.go's + // seedMaxBodyBytes), and nesting http.MaxBytesReader wraps only + // ever shrinks the effective limit to the smallest one in the + // chain, so this never fights those. It exists so a FUTURE route + // added here without remembering its own cap still gets bounded + // instead of silently inheriting an unlimited body. + r.Use(httputil.MaxBytes(8 << 20)) + // 64 KB — manifest + lifecycle install/upgrade/downgrade/uninstall + // bodies are tiny. Applied per-route via .With() rather than + // folded into the group backstop above so the seed route below + // (COPY-text chunks up to 2MB) can enforce its own larger cap + // instead of inheriting this smaller one — nesting two + // http.MaxBytesReader wraps only ever shrinks the effective limit + // to the smaller of the two. + smallBody := httputil.MaxBytes(64 * 1024) + r.With(smallBody).Get("/manifest", system.ManifestHandler( m.config.ID, m.config.Slug, m.config.Name, m.config.Icon, m.config.Tags, m.config.SQL, m.config.Versions, m.registry, m.contribReg, )) @@ -807,10 +819,29 @@ func (m *Module) mountSystemRoutes() { for _, scope := range migration.AllScopes() { runTx := m.lifecycleTxRunner(scope) r.Route("/"+string(scope), func(r chi.Router) { - r.Post("/install", system.InstallHandler(m.config.SQL, scope, runTx)) - r.Post("/upgrade", system.UpgradeHandler(m.config.SQL, scope, runTx)) - r.Post("/downgrade", system.DowngradeHandler(m.config.SQL, scope, runTx)) - r.Post("/uninstall", system.UninstallHandler()) // no scope — no-op for both + r.With(smallBody).Post("/install", system.InstallHandler(m.config.SQL, scope, runTx)) + r.With(smallBody).Post("/upgrade", system.UpgradeHandler(m.config.SQL, scope, runTx)) + r.With(smallBody).Post("/downgrade", system.DowngradeHandler(m.config.SQL, scope, runTx)) + r.With(smallBody).Post("/uninstall", system.UninstallHandler()) // no scope — no-op for both + + if scope == migration.ScopeApp && !runtime.IsLambda() { + // Dev-mount seed (F2 of the transient-overlay + // work): api-platform's devseed.Seeder POSTs one + // Postgres COPY-text chunk per call into this + // app's schema. App-scope only — there is no + // module-scope seed. SeedHandler enforces its own + // 2MB http.MaxBytesReader (system/lifecycle.go), + // so it deliberately does NOT get smallBody. + // + // !runtime.IsLambda() mirrors the lambda-invoke shim + // gate above (mountSystemRoutes, ~line 760): SeedHandler + // executes req.CreateSQL as raw request-body SQL, a + // surface the rest of the lifecycle deliberately + // withholds in prod. Dev/tunnel only. + r.Post("/seed", system.SeedHandler(func(ctx context.Context) (system.SeedConn, func(), error) { + return m.seedConn(ctx) + })) + } }) } }) diff --git a/internal/core/module_test.go b/internal/core/module_test.go index bb4b3b4..a052b2e 100644 --- a/internal/core/module_test.go +++ b/internal/core/module_test.go @@ -809,6 +809,59 @@ func TestLifecycle_ScopeTxRunnerWiring(t *testing.T) { } } +// TestSeed_RouteRequiresInternalSecret covers the dev-mount seed route's two +// gates: (1) it inherits internalAuth like every other /platform route, and +// (2) it is mounted only outside Lambda mode (mountSystemRoutes gates it with +// !runtime.IsLambda(), mirroring the lambda-invoke shim a few lines above it). +// +// The Lambda-mode branch (route absent entirely) can't be driven from this +// package the way TestLambdaInvokeShim_MountedPostOnly notes for the shim: +// runtime.IsLambda() reads an unexported package-level var evaluated once at +// process init from AWS_LAMBDA_FUNCTION_NAME, so a test in package core has +// no supported way to flip it mid-run. This test instead asserts the route is +// present (and secret-gated) in the non-Lambda/dev mount — the half the +// !runtime.IsLambda() guard in mountSystemRoutes is the reviewed line for. +func TestSeed_RouteRequiresInternalSecret(t *testing.T) { + m := newTestModuleWithSecret(t, "test") + route := "/__mirrorstack/platform/lifecycle/app/seed" + + // Without secret → 401, proving the route inherits internalAuth. + rec := doRequest(t, m.Router(), "POST", route) + if rec.Code != http.StatusUnauthorized { + t.Errorf("status without secret = %d, want 401", rec.Code) + } + + // With secret, in this (non-Lambda) test process, the route must be + // mounted — i.e. not 404. The handler itself will fail past auth (no + // real DB, malformed body) but that is out of scope for this test. + rec = doRequestWithSecret(t, m.Router(), "POST", route, "secret") + if rec.Code == http.StatusNotFound { + t.Errorf("status with secret = 404, want route to be mounted outside Lambda mode") + } +} + +// TestSeedRoute_MaxBytesLimit guards the /platform group's defense-in-depth +// backstop (mountSystemRoutes: r.Use(httputil.MaxBytes(8 << 20))) against +// regressing seed's own tighter cap. A body over seed's 2 MB reader +// (system/lifecycle.go's seedMaxBodyBytes) but well under the group's 8 MB +// backstop must still trip 413 from seed's own reader — proving nested +// http.MaxBytesReader wraps shrink to the smallest cap in the chain rather +// than the group backstop silently widening the effective limit to 8 MB. +func TestSeedRoute_MaxBytesLimit(t *testing.T) { + m := newTestModuleWithSecret(t, "test") + + huge := strings.Repeat("a", 2*1024*1024+1024) + body := `{"appId":"seedtest","table":"items","columns":["id"],"data":"` + huge + `","first":true}` + req := httptest.NewRequest("POST", "/__mirrorstack/platform/lifecycle/app/seed", strings.NewReader(body)) + req.Header.Set("X-MS-Internal-Secret", "secret") + rec := httptest.NewRecorder() + m.Router().ServeHTTP(rec, req) + + if rec.Code != http.StatusRequestEntityTooLarge { + t.Errorf("status = %d, want 413 for body over seed's 2MB cap, nested under the group's 8MB backstop", rec.Code) + } +} + func TestManifest_RegisteredScopesStillRouteCorrectly(t *testing.T) { // Verify that the chi.Walk + re-register approach in scopedRoutes preserves // the original routing behavior. Routes registered via Platform/Public/Internal diff --git a/internal/runtime/lambda.go b/internal/runtime/lambda.go index 68dce5a..995dbb0 100644 --- a/internal/runtime/lambda.go +++ b/internal/runtime/lambda.go @@ -22,6 +22,24 @@ import ( // validates it identically — keeping dev and prod from drifting on the shape. var AppSchemaPattern = regexp.MustCompile(`^app_[a-z0-9_]+$`) +// AppSchemaName derives the per-app Postgres schema from an app id, matching +// the platform's ids.AppSchemaName convention (app_). +// Returns ok=false if the result does not match AppSchemaPattern (e.g. appID +// is empty or contains characters outside [a-zA-Z0-9-]). +// +// Shared by every caller that must derive app_ from a bare app id rather +// than receive the schema pre-resolved: the dev per-request schema middleware +// (core.devAppSchemaMiddleware, via its devAppSchemaName wrapper) and the +// dev-mount seed endpoint (system.SeedHandler), whose wire contract carries +// only appId — see api-platform's devseed.Seeder. +func AppSchemaName(appID string) (string, bool) { + schema := "app_" + strings.ReplaceAll(strings.ToLower(appID), "-", "_") + if !AppSchemaPattern.MatchString(schema) { + return "", false + } + return schema, true +} + // Resources holds per-invocation credentials for all platform services. type Resources struct { DB *db.Credential `json:"db,omitempty"` diff --git a/system/lifecycle.go b/system/lifecycle.go index fc06a3c..78ada67 100644 --- a/system/lifecycle.go +++ b/system/lifecycle.go @@ -4,14 +4,19 @@ import ( "context" "encoding/json" "errors" + "fmt" "io" "io/fs" "net/http" "strconv" + "strings" + + "github.com/jackc/pgx/v5" "github.com/mirrorstack-ai/app-module-sdk/db" "github.com/mirrorstack-ai/app-module-sdk/internal/httputil" "github.com/mirrorstack-ai/app-module-sdk/internal/migration" + "github.com/mirrorstack-ai/app-module-sdk/internal/runtime" ) // LifecycleResult is the JSON response for install/upgrade/downgrade. Install @@ -302,3 +307,151 @@ func decodeUpgradeRequest(w http.ResponseWriter, r *http.Request) (UpgradeReques } return req, true } + +// seedMaxBodyBytes caps one seed POST's body. The platform's devseed.Seeder +// batches each COPY-text chunk to <=1MB (seedBatchBytes there); 2MB leaves +// headroom for the JSON envelope + column list without ever being the +// binding constraint on a well-behaved caller. +const seedMaxBodyBytes = 2 << 20 + +// SeedRequest is the platform->SDK wire shape for the dev-mount seed endpoint +// (devseed.Seeder on the api-platform side — a cross-repo contract; keep +// field tags byte-for-byte in sync with that package's seedRequest). One +// POST carries one Postgres COPY-text chunk for one table. +type SeedRequest struct { + AppID string `json:"appId"` + Table string `json:"table"` // physical m_* relation name + Columns []string `json:"columns"` // explicit COPY column list, ordinal order + CreateSQL string `json:"createSql,omitempty"` + Data string `json:"data"` // Postgres COPY text rows + First bool `json:"first"` // first chunk for this table +} + +// SeedResponse is the SDK's per-chunk answer. +type SeedResponse struct { + Skipped bool `json:"skipped,omitempty"` +} + +// SeedConn is the connection seam SeedHandler needs: the db.Querier surface +// for CreateSQL and the if-empty guard, plus the raw driver connection COPY +// FROM STDIN requires (db.Querier exposes neither PgConn() nor the +// low-level copy protocol). *pgxpool.Conn satisfies this directly — its +// Conn() method returns the *pgx.Conn that PgConn() rides on. +type SeedConn interface { + db.Querier + Conn() *pgx.Conn +} + +// SeedConnAcquirer resolves a SeedConn scoped to ctx's app schema for one +// request. Implementations must NOT route this through a cross-module dev +// guard the way Module.DB/Tx do: the seed endpoint legitimately writes into +// another module's exposure-anchored dependency tables, a grant the platform +// already enforced before it ever sent the chunk. +type SeedConnAcquirer func(ctx context.Context) (SeedConn, func(), error) + +// SeedHandler returns an http.HandlerFunc for the dev-mount seed endpoint: +// api-platform's devseed.Seeder POSTs one Postgres COPY-text chunk per call, +// targeting either the module's own table or an exposure-anchored producer +// dependency table (see package devseed on the platform side for the full +// design). This is the only lifecycle verb scoped to "app" alone — there is +// no module-scope seed, so the caller mounts it only under that scope's +// route (see internal/core/module.go's mountSystemRoutes). +// +// acquire resolves the connection for each request; it is responsible for +// putting ctx's app schema on search_path (via db.WithSchema + a scoped +// acquire, e.g. Module.seedConn) — SeedHandler itself only derives the +// schema NAME from RequestBody.AppID and injects it into ctx. +// +// First=true chunks run the create-then-if-empty sequence: CreateSQL (when +// non-empty) materializes a table the dev DB never migrated — a dependency +// table belonging to a producer module — then an if-empty check on Table +// decides whether to proceed. A non-empty table means the developer already +// has local data for it: the SDK answers {skipped:true} and does NOT touch +// the table, so a developer's local writes are never clobbered by a +// re-seed. First=false chunks (a continuation for a table already accepted +// on its first chunk) skip both steps and append unconditionally — the +// seeder stops sending further chunks the moment it sees {skipped:true} on +// the first one. +// +// Table and every entry of Columns are quoted via pgx.Identifier.Sanitize +// before use in any SQL. The platform derives them from its own pg_catalog +// read (see devseed.Seeder.columns), but the SDK quotes defensively rather +// than trust the wire. +func SeedHandler(acquire SeedConnAcquirer) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + r.Body = http.MaxBytesReader(w, r.Body, seedMaxBodyBytes) + + var req SeedRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeBodyDecodeError(w, err) + return + } + if req.Table == "" || len(req.Columns) == 0 { + httputil.JSON(w, http.StatusBadRequest, httputil.ErrorResponse{Error: "table and columns are required"}) + return + } + schema, ok := runtime.AppSchemaName(req.AppID) + if !ok { + httputil.JSON(w, http.StatusBadRequest, httputil.ErrorResponse{Error: "invalid appId"}) + return + } + + ctx := db.WithSchema(r.Context(), schema) + conn, release, err := acquire(ctx) + if err != nil { + httputil.JSON(w, http.StatusInternalServerError, httputil.ErrorResponse{Error: "acquire db connection: " + err.Error()}) + return + } + defer release() + + quotedTable := pgx.Identifier{req.Table}.Sanitize() + + if req.First { + if req.CreateSQL != "" { + if _, err := conn.Exec(ctx, req.CreateSQL); err != nil { + httputil.JSON(w, http.StatusInternalServerError, httputil.ErrorResponse{Error: "create table: " + err.Error()}) + return + } + } + empty, err := seedTableIsEmpty(ctx, conn, quotedTable) + if err != nil { + httputil.JSON(w, http.StatusInternalServerError, httputil.ErrorResponse{Error: "check table empty: " + err.Error()}) + return + } + if !empty { + // The developer's own local writes win — drop this and every + // remaining chunk for the table (the seeder stops sending on + // {skipped:true}) without touching a single row. + httputil.JSON(w, http.StatusOK, SeedResponse{Skipped: true}) + return + } + } + + if req.Data != "" { + quotedCols := make([]string, len(req.Columns)) + for i, c := range req.Columns { + quotedCols[i] = pgx.Identifier{c}.Sanitize() + } + copySQL := fmt.Sprintf("COPY %s (%s) FROM STDIN", quotedTable, strings.Join(quotedCols, ", ")) + if _, err := conn.Conn().PgConn().CopyFrom(ctx, strings.NewReader(req.Data), copySQL); err != nil { + httputil.JSON(w, http.StatusInternalServerError, httputil.ErrorResponse{Error: "copy data: " + err.Error()}) + return + } + } + // Data == "" is valid: a First chunk that only needed CreateSQL to + // materialize an empty dependency table (see devseed.Seeder.seedTable + // on the platform side, the tail case for an empty own-table skip). + + httputil.JSON(w, http.StatusOK, SeedResponse{}) + } +} + +// seedTableIsEmpty reports whether quotedTable (already identifier-quoted) +// has any row, scoped by conn's current search_path. +func seedTableIsEmpty(ctx context.Context, conn SeedConn, quotedTable string) (bool, error) { + var exists bool + if err := conn.QueryRow(ctx, "SELECT EXISTS (SELECT 1 FROM "+quotedTable+" LIMIT 1)").Scan(&exists); err != nil { + return false, err + } + return !exists, nil +} diff --git a/system/seed_integration_test.go b/system/seed_integration_test.go new file mode 100644 index 0000000..18bc1df --- /dev/null +++ b/system/seed_integration_test.go @@ -0,0 +1,271 @@ +//go:build integration + +package system + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/mirrorstack-ai/app-module-sdk/db" +) + +// seedTestDB opens the dev pool for the seed integration tests, skipping +// (not failing) when no Postgres is reachable — same convention as +// db.testDB / core's dependency_db_deployed_integration_test.go. +func seedTestDB(t *testing.T) *db.DB { + t.Helper() + d, err := db.Open(context.Background()) + if err != nil { + t.Skipf("skipping: cannot connect to postgres: %v", err) + } + t.Cleanup(d.Close) + return d +} + +// seedAcquirerFor wires a SeedConnAcquirer straight to db.AcquireScopedConn — +// the same acquire path Module.seedConn uses in internal/core/db.go, minus +// the pool-cache/dev-guard machinery that package doesn't need to prove +// here. SeedHandler itself is responsible for putting the app schema on ctx +// before calling this. +func seedAcquirerFor(d *db.DB) SeedConnAcquirer { + return func(ctx context.Context) (SeedConn, func(), error) { + return db.AcquireScopedConn(ctx, d.Pool()) + } +} + +// doSeed POSTs req through SeedHandler and decodes the response. +func doSeed(t *testing.T, h http.HandlerFunc, req SeedRequest) (int, SeedResponse) { + t.Helper() + body, err := json.Marshal(req) + if err != nil { + t.Fatalf("marshal request: %v", err) + } + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest("POST", "/seed", strings.NewReader(string(body)))) + + var resp SeedResponse + if rec.Code == http.StatusOK { + if err := json.Unmarshal(rec.Body.Bytes(), &resp); err != nil { + t.Fatalf("decode response: %v; body=%s", err, rec.Body.String()) + } + } + return rec.Code, resp +} + +func TestSeedHandler_Integration_FirstChunkIntoEmptyTable(t *testing.T) { + d := seedTestDB(t) + ctx := context.Background() + + const appID = "seedtestempty" + const schema = "app_seedtestempty" + mustExecAdmin(t, d, ctx, "DROP SCHEMA IF EXISTS "+schema+" CASCADE") + mustExecAdmin(t, d, ctx, "CREATE SCHEMA "+schema) + t.Cleanup(func() { d.Exec(ctx, "DROP SCHEMA IF EXISTS "+schema+" CASCADE") }) + // Already-migrated own table: pre-existing and empty, the common case for + // a freshly dev-provisioned app. + mustExecAdmin(t, d, ctx, "CREATE TABLE "+schema+".items (id int, title text)") + + h := SeedHandler(seedAcquirerFor(d)) + code, resp := doSeed(t, h, SeedRequest{ + AppID: appID, + Table: "items", + Columns: []string{"id", "title"}, + Data: "1\tfoo\n2\tbar\n", + First: true, + }) + if code != http.StatusOK { + t.Fatalf("status = %d, want 200", code) + } + if resp.Skipped { + t.Fatalf("Skipped = true, want false (table was empty)") + } + + rows := queryTitles(t, d, ctx, schema, "items") + want := map[string]bool{"foo": true, "bar": true} + if len(rows) != 2 || !want[rows[0]] || !want[rows[1]] { + t.Errorf("titles = %v, want exactly [foo bar] in some order", rows) + } +} + +func TestSeedHandler_Integration_FirstChunkIntoNonEmptyTable(t *testing.T) { + d := seedTestDB(t) + ctx := context.Background() + + const appID = "seedtestnonempty" + const schema = "app_seedtestnonempty" + mustExecAdmin(t, d, ctx, "DROP SCHEMA IF EXISTS "+schema+" CASCADE") + mustExecAdmin(t, d, ctx, "CREATE SCHEMA "+schema) + t.Cleanup(func() { d.Exec(ctx, "DROP SCHEMA IF EXISTS "+schema+" CASCADE") }) + mustExecAdmin(t, d, ctx, "CREATE TABLE "+schema+".items (id int, title text)") + // The developer's own local write — must survive the seed untouched. + mustExecAdmin(t, d, ctx, "INSERT INTO "+schema+".items (id, title) VALUES (99, 'developer-local')") + + h := SeedHandler(seedAcquirerFor(d)) + code, resp := doSeed(t, h, SeedRequest{ + AppID: appID, + Table: "items", + Columns: []string{"id", "title"}, + Data: "1\tfoo\n", + First: true, + }) + if code != http.StatusOK { + t.Fatalf("status = %d, want 200", code) + } + if !resp.Skipped { + t.Fatalf("Skipped = false, want true (table already had a row)") + } + + rows := queryTitles(t, d, ctx, schema, "items") + if len(rows) != 1 || rows[0] != "developer-local" { + t.Errorf("titles = %v, want exactly [developer-local] — seed must not touch a non-empty table", rows) + } +} + +func TestSeedHandler_Integration_CreateSQLDependencyTable(t *testing.T) { + d := seedTestDB(t) + ctx := context.Background() + + const appID = "seedtestdep" + const schema = "app_seedtestdep" + mustExecAdmin(t, d, ctx, "DROP SCHEMA IF EXISTS "+schema+" CASCADE") + mustExecAdmin(t, d, ctx, "CREATE SCHEMA "+schema) + t.Cleanup(func() { d.Exec(ctx, "DROP SCHEMA IF EXISTS "+schema+" CASCADE") }) + // Deliberately no pre-existing table: this is the dependency-table case — + // the dev DB never migrated a producer module it doesn't own. + + h := SeedHandler(seedAcquirerFor(d)) + code, resp := doSeed(t, h, SeedRequest{ + AppID: appID, + Table: "m81b3ac7081c1409495700c761e23b59e_categories", + Columns: []string{"id", "title"}, + CreateSQL: `CREATE TABLE IF NOT EXISTS "m81b3ac7081c1409495700c761e23b59e_categories" (id int, title text)`, + Data: "1\tproducer-row\n", + First: true, + }) + if code != http.StatusOK { + t.Fatalf("status = %d, want 200; body should carry {skipped:false}", code) + } + if resp.Skipped { + t.Fatalf("Skipped = true, want false (dependency table was freshly created and empty)") + } + + rows := queryTitles(t, d, ctx, schema, `"m81b3ac7081c1409495700c761e23b59e_categories"`) + if len(rows) != 1 || rows[0] != "producer-row" { + t.Errorf("titles = %v, want exactly [producer-row]", rows) + } +} + +func TestSeedHandler_Integration_CreateSQLOnlyEmptyDependencyTable(t *testing.T) { + d := seedTestDB(t) + ctx := context.Background() + + const appID = "seedtestdepempty" + const schema = "app_seedtestdepempty" + mustExecAdmin(t, d, ctx, "DROP SCHEMA IF EXISTS "+schema+" CASCADE") + mustExecAdmin(t, d, ctx, "CREATE SCHEMA "+schema) + t.Cleanup(func() { d.Exec(ctx, "DROP SCHEMA IF EXISTS "+schema+" CASCADE") }) + + // A producer's dependency table with zero live rows still needs the + // CREATE materialized locally — the seeder sends CreateSQL with no Data + // (see devseed.Seeder.seedTable's tail case on the platform side). + h := SeedHandler(seedAcquirerFor(d)) + code, resp := doSeed(t, h, SeedRequest{ + AppID: appID, + Table: "m81b3ac7081c1409495700c761e23b59e_empty_dep", + Columns: []string{"id"}, + CreateSQL: `CREATE TABLE IF NOT EXISTS "m81b3ac7081c1409495700c761e23b59e_empty_dep" (id int)`, + Data: "", + First: true, + }) + if code != http.StatusOK { + t.Fatalf("status = %d, want 200", code) + } + if resp.Skipped { + t.Fatalf("Skipped = true, want false") + } + + var exists bool + if err := d.Pool().QueryRow(ctx, + `SELECT EXISTS (SELECT 1 FROM information_schema.tables WHERE table_schema = $1 AND table_name = $2)`, + schema, "m81b3ac7081c1409495700c761e23b59e_empty_dep").Scan(&exists); err != nil { + t.Fatalf("check table exists: %v", err) + } + if !exists { + t.Error("dependency table was not created") + } +} + +func TestSeedHandler_Integration_ContinuationChunkAppends(t *testing.T) { + d := seedTestDB(t) + ctx := context.Background() + + const appID = "seedtestcontinuation" + const schema = "app_seedtestcontinuation" + mustExecAdmin(t, d, ctx, "DROP SCHEMA IF EXISTS "+schema+" CASCADE") + mustExecAdmin(t, d, ctx, "CREATE SCHEMA "+schema) + t.Cleanup(func() { d.Exec(ctx, "DROP SCHEMA IF EXISTS "+schema+" CASCADE") }) + mustExecAdmin(t, d, ctx, "CREATE TABLE "+schema+".items (id int, title text)") + + h := SeedHandler(seedAcquirerFor(d)) + code, resp := doSeed(t, h, SeedRequest{ + AppID: appID, Table: "items", Columns: []string{"id", "title"}, + Data: "1\tfoo\n", First: true, + }) + if code != http.StatusOK || resp.Skipped { + t.Fatalf("first chunk: status=%d skipped=%v, want 200/false", code, resp.Skipped) + } + // Continuation chunk: First=false must skip the if-empty guard (the table + // now has a row from the chunk above) and append rather than replace. + code, resp = doSeed(t, h, SeedRequest{ + AppID: appID, Table: "items", Columns: []string{"id", "title"}, + Data: "2\tbar\n", First: false, + }) + if code != http.StatusOK || resp.Skipped { + t.Fatalf("continuation chunk: status=%d skipped=%v, want 200/false", code, resp.Skipped) + } + + rows := queryTitles(t, d, ctx, schema, "items") + if len(rows) != 2 { + t.Errorf("titles = %v, want 2 rows (foo appended with bar, not replaced)", rows) + } +} + +// mustExecAdmin runs sql on the unscoped dev pool (no app schema on ctx) — +// used for test fixture setup/teardown, which operates on schema-qualified +// names directly rather than through the app-schema search_path SeedHandler +// itself relies on. +func mustExecAdmin(t *testing.T, d *db.DB, ctx context.Context, sql string) { + t.Helper() + if _, err := d.Exec(ctx, sql); err != nil { + t.Fatalf("mustExecAdmin %q: %v", sql, err) + } +} + +// queryTitles reads every "title"-ish value back out of schema.table in +// insertion order, for asserting COPY landed the expected rows. table may +// already be identifier-quoted by the caller (for m_ names that need +// exact-case quoting); schema is always plain and gets quoted here. +func queryTitles(t *testing.T, d *db.DB, ctx context.Context, schema, table string) []string { + t.Helper() + rows, err := d.Pool().Query(ctx, `SELECT title FROM `+schema+"."+table) + if err != nil { + t.Fatalf("query %s.%s: %v", schema, table, err) + } + defer rows.Close() + var out []string + for rows.Next() { + var title string + if err := rows.Scan(&title); err != nil { + t.Fatalf("scan: %v", err) + } + out = append(out, title) + } + if err := rows.Err(); err != nil { + t.Fatalf("rows: %v", err) + } + return out +} diff --git a/system/seed_test.go b/system/seed_test.go new file mode 100644 index 0000000..61511e2 --- /dev/null +++ b/system/seed_test.go @@ -0,0 +1,137 @@ +package system + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +// noopSeedAcquirer is a SeedConnAcquirer stub that fails the test if invoked. +// Mirrors noopTxRunner in lifecycle_test.go: tests using it assert that +// SeedHandler rejects the request BEFORE ever trying to reach the database. +func noopSeedAcquirer(t *testing.T) SeedConnAcquirer { + return func(ctx context.Context) (SeedConn, func(), error) { + t.Helper() + t.Errorf("SeedConnAcquirer unexpectedly called — test assumed the handler would reject before acquiring a connection") + return nil, nil, nil + } +} + +func TestSeedHandler_MalformedBody_400(t *testing.T) { + t.Parallel() + + h := SeedHandler(noopSeedAcquirer(t)) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest("POST", "/seed", strings.NewReader(`{not json`))) + + if rec.Code != http.StatusBadRequest { + t.Errorf("status = %d, want 400 for malformed body", rec.Code) + } +} + +func TestSeedHandler_EmptyBody_400(t *testing.T) { + t.Parallel() + + h := SeedHandler(noopSeedAcquirer(t)) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest("POST", "/seed", nil)) + + // Unlike install (whose body is entirely optional), every seed field is + // required — an empty body decodes to io.EOF, which writeBodyDecodeError + // maps to 400 same as any other malformed body. + if rec.Code != http.StatusBadRequest { + t.Errorf("status = %d, want 400 for empty body", rec.Code) + } +} + +func TestSeedHandler_OversizeBody_413(t *testing.T) { + t.Parallel() + + // A single "data" field bigger than seedMaxBodyBytes must trip + // http.MaxBytesReader before the JSON decoder ever finishes — the + // acquirer must never be reached. + huge := strings.Repeat("a", seedMaxBodyBytes+1024) + body := `{"appId":"seedtest","table":"items","columns":["id"],"data":"` + huge + `","first":true}` + + h := SeedHandler(noopSeedAcquirer(t)) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest("POST", "/seed", strings.NewReader(body))) + + if rec.Code != http.StatusRequestEntityTooLarge { + t.Errorf("status = %d, want 413 for oversize body", rec.Code) + } +} + +func TestSeedHandler_MissingTableOrColumns_400(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + body string + }{ + {"missing table", `{"appId":"seedtest","columns":["id"],"data":"1\n","first":true}`}, + {"empty table", `{"appId":"seedtest","table":"","columns":["id"],"data":"1\n","first":true}`}, + {"missing columns", `{"appId":"seedtest","table":"items","data":"1\n","first":true}`}, + {"empty columns", `{"appId":"seedtest","table":"items","columns":[],"data":"1\n","first":true}`}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + h := SeedHandler(noopSeedAcquirer(t)) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest("POST", "/seed", strings.NewReader(tc.body))) + + if rec.Code != http.StatusBadRequest { + t.Errorf("status = %d, want 400 for body %q", rec.Code, tc.body) + } + }) + } +} + +func TestSeedHandler_InvalidAppID_400(t *testing.T) { + t.Parallel() + + cases := []struct { + name string + body string + }{ + {"empty appId", `{"appId":"","table":"items","columns":["id"],"data":"1\n","first":true}`}, + {"appId with invalid chars", `{"appId":"not an id!","table":"items","columns":["id"],"data":"1\n","first":true}`}, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + h := SeedHandler(noopSeedAcquirer(t)) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest("POST", "/seed", strings.NewReader(tc.body))) + + if rec.Code != http.StatusBadRequest { + t.Errorf("status = %d, want 400 for body %q", rec.Code, tc.body) + } + }) + } +} + +func TestSeedHandler_AcquireError_500(t *testing.T) { + t.Parallel() + + acquireErr := errAcquire{} + h := SeedHandler(func(ctx context.Context) (SeedConn, func(), error) { + return nil, nil, acquireErr + }) + body := `{"appId":"seedtest","table":"items","columns":["id"],"data":"1\n","first":true}` + rec := httptest.NewRecorder() + h.ServeHTTP(rec, httptest.NewRequest("POST", "/seed", strings.NewReader(body))) + + if rec.Code != http.StatusInternalServerError { + t.Errorf("status = %d, want 500 when acquire fails", rec.Code) + } +} + +type errAcquire struct{} + +func (errAcquire) Error() string { return "acquire failed" }