← back to Cli Printing Press
feat(cli): sync.walker spec parameter for hierarchical APIs (#1060) (#1074)
24962b7c862f5f4d6cb3679908615266d4c531e4 · 2026-05-11 10:14:46 -0700 · Justin Fu
Adds Tier 2 of the hierarchical-sync work — a spec-declared `walker:`
parameter (internal YAML) / `x-pp-sync-walker` (OpenAPI operation
extension) that lets spec authors declare parent → child sync chains
that the profiler's path-param auto-detection would otherwise miss,
including non-PK key extraction from the parent record.
Scope follows the issue boundary: 2-level parent → child only;
deeper trees deferred.
Builds on Tier 1 (#1006 / PR #1009) — the unresolvedPathKeyRE skip
block is the safety net; walker-declared endpoints flow through the
existing dependent-sync machinery instead of being skipped.
Surfaces:
- spec.WalkerConfig{Parent, KeyField, KeyParam} on Endpoint
- OpenAPI x-pp-sync-walker extension on operations
- profiler.applySpecWalkers synthesizes or augments DependentResource
- DependentResource gains KeyField (non-PK extraction)
- sync.go.tmpl: dependentResourceDef.KeyField + branch on it
- store.go.tmpl: new ListField(table, field) helper (symmetric to ListIDs)
- New golden fixture testdata/golden/fixtures/sync-walker-api.yaml
- 4 profiler tests + 1 OpenAPI test + 1 YAML round-trip test
- docs/SPEC-EXTENSIONS.md updated
Walkers pointing at non-syncable parents emit a stderr warning and are
dropped — explicit author intent must not silently produce passing
builds with missing data. Multi-placeholder paths without explicit
key_param also warn-and-drop to avoid firstPathParam's "first wins"
default picking the parent slot on a 2-deep path.
Goldens regenerated: generate-golden-api/sync.go diffs by KeyField
addition; generate-sync-walker-api is a new case.
Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Files touched
M docs/SPEC-EXTENSIONS.mdM internal/generator/generator.goM internal/generator/templates/store.go.tmplM internal/generator/templates/sync.go.tmplM internal/openapi/parser.goM internal/openapi/parser_test.goM internal/profiler/profiler.goM internal/profiler/profiler_test.goM internal/spec/spec.goM internal/spec/spec_test.goA testdata/golden/cases/generate-sync-walker-api/artifacts.txtA testdata/golden/cases/generate-sync-walker-api/command.txtM testdata/golden/expected/generate-golden-api/printing-press-golden/internal/cli/sync.goA testdata/golden/expected/generate-sync-walker-api/exit.txtA testdata/golden/expected/generate-sync-walker-api/stderr.txtA testdata/golden/expected/generate-sync-walker-api/stdout.txtA testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/.printing-press.jsonA testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/cli/promoted_games.goA testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/cli/promoted_leagues.goA testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/cli/sync.goA testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/store/store.goA testdata/golden/fixtures/sync-walker-api.yaml
Diff
commit 24962b7c862f5f4d6cb3679908615266d4c531e4
Author: Justin Fu <justinwfu@gmail.com>
Date: Mon May 11 10:14:46 2026 -0700
feat(cli): sync.walker spec parameter for hierarchical APIs (#1060) (#1074)
Adds Tier 2 of the hierarchical-sync work — a spec-declared `walker:`
parameter (internal YAML) / `x-pp-sync-walker` (OpenAPI operation
extension) that lets spec authors declare parent → child sync chains
that the profiler's path-param auto-detection would otherwise miss,
including non-PK key extraction from the parent record.
Scope follows the issue boundary: 2-level parent → child only;
deeper trees deferred.
Builds on Tier 1 (#1006 / PR #1009) — the unresolvedPathKeyRE skip
block is the safety net; walker-declared endpoints flow through the
existing dependent-sync machinery instead of being skipped.
Surfaces:
- spec.WalkerConfig{Parent, KeyField, KeyParam} on Endpoint
- OpenAPI x-pp-sync-walker extension on operations
- profiler.applySpecWalkers synthesizes or augments DependentResource
- DependentResource gains KeyField (non-PK extraction)
- sync.go.tmpl: dependentResourceDef.KeyField + branch on it
- store.go.tmpl: new ListField(table, field) helper (symmetric to ListIDs)
- New golden fixture testdata/golden/fixtures/sync-walker-api.yaml
- 4 profiler tests + 1 OpenAPI test + 1 YAML round-trip test
- docs/SPEC-EXTENSIONS.md updated
Walkers pointing at non-syncable parents emit a stderr warning and are
dropped — explicit author intent must not silently produce passing
builds with missing data. Multi-placeholder paths without explicit
key_param also warn-and-drop to avoid firstPathParam's "first wins"
default picking the parent slot on a 2-deep path.
Goldens regenerated: generate-golden-api/sync.go diffs by KeyField
addition; generate-sync-walker-api is a new case.
Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
---
docs/SPEC-EXTENSIONS.md | 72 ++
internal/generator/generator.go | 7 +
internal/generator/templates/store.go.tmpl | 80 ++
internal/generator/templates/sync.go.tmpl | 23 +-
internal/openapi/parser.go | 33 +
internal/openapi/parser_test.go | 74 ++
internal/profiler/profiler.go | 142 +++
internal/profiler/profiler_test.go | 276 +++++
internal/spec/spec.go | 33 +-
internal/spec/spec_test.go | 55 +
.../cases/generate-sync-walker-api/artifacts.txt | 5 +
.../cases/generate-sync-walker-api/command.txt | 2 +
.../printing-press-golden/internal/cli/sync.go | 23 +-
.../expected/generate-sync-walker-api/exit.txt | 1 +
.../expected/generate-sync-walker-api/stderr.txt | 2 +
.../expected/generate-sync-walker-api/stdout.txt | 0
.../sync-walker-golden/.printing-press.json | 23 +
.../internal/cli/promoted_games.go | 84 ++
.../internal/cli/promoted_leagues.go | 98 ++
.../sync-walker-golden/internal/cli/sync.go | 1247 ++++++++++++++++++++
.../sync-walker-golden/internal/store/store.go | 1150 ++++++++++++++++++
testdata/golden/fixtures/sync-walker-api.yaml | 44 +
22 files changed, 3466 insertions(+), 8 deletions(-)
diff --git a/docs/SPEC-EXTENSIONS.md b/docs/SPEC-EXTENSIONS.md
index 517a2148..b534f694 100644
--- a/docs/SPEC-EXTENSIONS.md
+++ b/docs/SPEC-EXTENSIONS.md
@@ -34,6 +34,7 @@ in the same change as any new `Extensions["x-*"]` lookup in that file.
| `x-resource-id` | path item | `Endpoint.IDField` | No |
| `x-critical` | path item | `Endpoint.Critical` | No |
| `x-tier` | path item or operation | `Endpoint.Tier` | No |
+| `x-pp-sync-walker` | operation | `Endpoint.Walker` | No |
## `info` Extensions
@@ -591,3 +592,74 @@ paths:
responses:
"200": {description: ok}
```
+
+### `x-pp-sync-walker`
+
+Declares a hierarchical-walk dependency for a child endpoint. Synthesizes (or
+augments) a dependent-resource entry so the generator's existing
+parent-child sync machinery handles the fan-out — fetch the parent, extract
+the named field from each parent record, substitute it into the child path,
+fetch each child.
+
+Use this when the auto-detected parent-child link in the profiler would miss
+your endpoint or pick the wrong parent. Common cases:
+
+- The child path's placeholder name does not match a parent resource (e.g.
+ `/games/{game_key}/leagues` — `game_key` does not stem to "games" via the
+ default `_id`/`_key` stripping).
+- The parent placeholder lives in a matrix or query parameter rather than the
+ path, so the path has no `{placeholder}` for auto-detection to read.
+- The child path uses a parent field that is not the parent's primary key
+ (e.g. Yahoo Fantasy's `game_key`, Reddit's `subreddit` name).
+
+Parsed field: `Endpoint.Walker` (a `*spec.WalkerConfig`)
+
+Rules:
+- Optional.
+- Operation-level only. (No path-item-level form today.)
+- `parent` (string, required): the resource name to iterate. The parent must
+ itself be a syncable resource (i.e., have a flat-list endpoint). Walkers
+ pointing at non-syncable parents emit a `warning:` to stderr at generate
+ time and are dropped.
+- `key_field` (string, optional): the field to extract from each parent
+ record for substitution into the child path. Defaults to the parent's
+ primary key. Set this when the child path needs a non-PK field.
+- `key_param` (string, optional): the placeholder name in the child path
+ that receives the extracted value. Defaults to the first (and only)
+ `{placeholder}` in the child path when there is exactly one. **Required
+ explicitly when the child path has 0 or 2+ placeholders** — the
+ single-placeholder default would otherwise pick the wrong slot (or no
+ slot at all). The generator warns and drops the walker when it's ambiguous
+ and `key_param` is missing.
+- Walker-emitted dependents flow through the same `syncDependentResource`
+ machinery as auto-detected ones, so concurrency/retry/cursor/Upsert
+ behavior is identical.
+
+Internal YAML emits this as `walker:` on the endpoint with the same
+sub-field names (`parent`, `key_field`, `key_param`). Both surfaces parse
+to the same `WalkerConfig` struct.
+
+Example:
+
+```yaml
+paths:
+ /games:
+ get:
+ summary: List games (parent for the walker below)
+ responses:
+ "200": {description: ok}
+ /games/{game_key}/leagues:
+ get:
+ summary: List leagues for a game
+ x-pp-sync-walker:
+ parent: games
+ key_field: game_key
+ key_param: game_key
+ parameters:
+ - name: game_key
+ in: path
+ required: true
+ schema: {type: string}
+ responses:
+ "200": {description: ok}
+```
diff --git a/internal/generator/generator.go b/internal/generator/generator.go
index 342b2b37..dbf7c8e7 100644
--- a/internal/generator/generator.go
+++ b/internal/generator/generator.go
@@ -1947,6 +1947,13 @@ func (g *Generator) renderVisionAndRootFiles(promotedCommands []PromotedCommand,
return g.renderRootProjectFiles(promotedCommands, promotedResourceNames, workflowConstructors, insightConstructors)
}
+// schemaWithDependentParents adds a parent_id column + index to every
+// dependent resource's table so sync can record which parent each row
+// belongs to. For walker-emitted dependents whose DependentResource.KeyField
+// is non-empty, parent_id stores the value of that field (not strictly a
+// parent's primary key); the column name is retained for backwards
+// compatibility with existing CLIs. The naming caveat is internal — the
+// column is not part of any user-visible API.
func (g *Generator) schemaWithDependentParents() []TableDef {
schema := BuildSchema(g.Spec)
diff --git a/internal/generator/templates/store.go.tmpl b/internal/generator/templates/store.go.tmpl
index e86553d7..232ae72c 100644
--- a/internal/generator/templates/store.go.tmpl
+++ b/internal/generator/templates/store.go.tmpl
@@ -23,6 +23,13 @@ import (
var uuidPattern = regexp.MustCompile(`^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$`)
+// validIdentifierRE pins ListField's `field` argument to a safe SQL
+// identifier shape before any Sprintf interpolation. Matches what
+// pragma_table_info implicitly enforces on the primary path, so the
+// fallback path inherits the same defense without depending on whether
+// the parent's typed domain table exists at the moment of the lookup.
+var validIdentifierRE = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`)
+
// IsUUID returns true if the input looks like a UUID.
func IsUUID(s string) bool {
return uuidPattern.MatchString(s)
@@ -1075,6 +1082,79 @@ func (s *Store) ListIDs(resourceType string) ([]string, error) {
return ids, rows.Err()
}
+// ListField returns values of a named field from a resource's domain table,
+// or from the generic resources table via json_extract when no typed column
+// exists. Used by dependent sync to iterate parents when a spec-declared
+// walker extracts a non-PK field (Endpoint.Walker.KeyField in the upstream
+// printing-press repo) for the child path's placeholder.
+//
+// Defense in depth: field is validated against validIdentifierRE at entry
+// — the regex pins it to SQL-safe identifier shape covering both the
+// typed-column primary path AND the json_extract fallback (where
+// pragma_table_info validation would never run if the parent's domain
+// table doesn't exist yet). resourceType is never interpolated into SQL
+// directly; we resolve it to a real table name via a parameterized
+// sqlite_master lookup. Only validated names are substituted
+// (double-quoted) into the SELECT. Mirrors ListIDs's defense pattern so
+// callers may pass any string.
+func (s *Store) ListField(resourceType, field string) ([]string, error) {
+ if !validIdentifierRE.MatchString(field) {
+ return nil, fmt.Errorf("ListField: invalid field name %q (must match %s)", field, validIdentifierRE.String())
+ }
+ var table string
+ err := s.db.QueryRow(
+ `SELECT name FROM sqlite_master WHERE type='table' AND name=?`,
+ resourceType,
+ ).Scan(&table)
+ var rows *sql.Rows
+ if err == nil && table != "" {
+ // Validate the column exists on the resolved table before splicing
+ // it into the SELECT. pragma_table_info is parameterizable.
+ var colName string
+ colErr := s.db.QueryRow(
+ `SELECT name FROM pragma_table_info(?) WHERE name=?`,
+ table, field,
+ ).Scan(&colName)
+ if colErr == nil && colName != "" {
+ qTable := strings.ReplaceAll(table, `"`, `""`)
+ qCol := strings.ReplaceAll(colName, `"`, `""`)
+ // DISTINCT: callers iterate the returned values as parent keys
+ // for child-resource fan-out. Multiple parent rows sharing a
+ // key_field value (legal for non-PK fields) would otherwise
+ // cause the child endpoint to be fetched once per duplicate row.
+ rows, err = s.db.Query(fmt.Sprintf(
+ `SELECT DISTINCT "%s" FROM "%s" WHERE "%s" IS NOT NULL AND "%s" != ''`,
+ qCol, qTable, qCol, qCol,
+ ))
+ } else {
+ err = colErr
+ }
+ }
+ if err != nil || rows == nil {
+ // Fall back to generic resources table via json_extract. Path is
+ // Sprintf'd into the SQL string (matches ResolveByName below).
+ // DISTINCT for the same reason as the typed-column path above.
+ fallback := fmt.Sprintf(
+ `SELECT DISTINCT json_extract(data, '$.%s') FROM resources WHERE resource_type = ? AND json_extract(data, '$.%s') IS NOT NULL`,
+ field, field,
+ )
+ rows, err = s.db.Query(fallback, resourceType)
+ if err != nil {
+ return nil, err
+ }
+ }
+ defer rows.Close()
+
+ var values []string
+ for rows.Next() {
+ var v sql.NullString
+ if err := rows.Scan(&v); err == nil && v.Valid && v.String != "" {
+ values = append(values, v.String)
+ }
+ }
+ return values, rows.Err()
+}
+
// GetLastSyncedAt returns the last sync timestamp for a resource type.
func (s *Store) GetLastSyncedAt(resourceType string) string {
var ts sql.NullString
diff --git a/internal/generator/templates/sync.go.tmpl b/internal/generator/templates/sync.go.tmpl
index 25e5dd6f..83e44547 100644
--- a/internal/generator/templates/sync.go.tmpl
+++ b/internal/generator/templates/sync.go.tmpl
@@ -1046,17 +1046,23 @@ func syncClientForResource(c *client.Client, resource string) *client.Client {
{{- if .DependentSyncResources}}
// dependentResourceDef describes a child resource that requires iterating parent IDs to sync.
+// When KeyField is non-empty, the dependent's parent IDs are extracted from the parent
+// records' KeyField via json_extract rather than from the parent table's primary key.
+// Populated from a spec-declared walker (Endpoint.Walker.KeyField in internal YAML,
+// or `key_field` under `x-pp-sync-walker` in OpenAPI). Empty KeyField preserves the
+// existing parent-primary-key flow byte-for-byte.
type dependentResourceDef struct {
Name string
ParentTable string
ParentIDParam string
PathTemplate string
+ KeyField string
}
func dependentResourceDefs() []dependentResourceDef {
return []dependentResourceDef{
{{- range .DependentSyncResources}}
- {Name: "{{.Name}}", ParentTable: "{{.ParentResource}}", ParentIDParam: "{{.ParentIDParam}}", PathTemplate: "{{.Path}}"},
+ {Name: "{{.Name}}", ParentTable: "{{.ParentResource}}", ParentIDParam: "{{.ParentIDParam}}", PathTemplate: "{{.Path}}", KeyField: "{{.KeyField}}"},
{{- end}}
}
}
@@ -1090,8 +1096,19 @@ func syncDependentResource(c interface {
}, db *store.Store, dep dependentResourceDef, sinceTS string, full bool, maxPages int{{if .Pagination.DateRangeParam}}, dates string{{end}}) syncResult {
started := time.Now()
- // Query parent table for all IDs
- parentIDs, err := db.ListIDs(dep.ParentTable)
+ // Query parent table for the keys to substitute into the child path.
+ // When KeyField is empty, use the parent's primary key via ListIDs
+ // (the original flat parent-child flow). When KeyField is set, the
+ // spec declared a walker that extracts a non-PK field from each parent
+ // record — ListField looks up the field in the parent's typed column
+ // if present, otherwise json_extract from the generic resources table.
+ var parentIDs []string
+ var err error
+ if dep.KeyField != "" {
+ parentIDs, err = db.ListField(dep.ParentTable, dep.KeyField)
+ } else {
+ parentIDs, err = db.ListIDs(dep.ParentTable)
+ }
if err != nil || len(parentIDs) == 0 {
if len(parentIDs) == 0 {
if humanFriendly {
diff --git a/internal/openapi/parser.go b/internal/openapi/parser.go
index 4fada5ed..2e800e40 100644
--- a/internal/openapi/parser.go
+++ b/internal/openapi/parser.go
@@ -42,6 +42,7 @@ const (
extensionTierRouting = "x-tier-routing"
extensionTier = "x-tier"
extensionMCP = "x-mcp"
+ extensionSyncWalker = "x-pp-sync-walker"
extensionAPIName = "x-api-name"
extensionDisplayName = "x-display-name"
extensionWebsite = "x-website"
@@ -1825,6 +1826,7 @@ func mapResources(doc *openapi3.T, out *spec.APISpec, basePath string) {
endpoint.IDField = resolveIDFieldFromResponseSchema(op, targetResourceName)
}
endpoint.Critical = pathCritical
+ endpoint.Walker = readWalkerExtension(op.Extensions, fmt.Sprintf("%s %q", strings.ToUpper(method), path))
targetEndpoints[endpointName] = endpoint
@@ -2938,6 +2940,37 @@ func readTierExtension(extensions map[string]any, context string) string {
return strings.TrimSpace(tier)
}
+// readWalkerExtension reads the `x-pp-sync-walker` extension from an
+// operation's Extensions map and returns a parsed WalkerConfig. The raw
+// extension value is expected to be a JSON/YAML object with `parent`,
+// `key_field`, and `key_param` keys; missing keys default per the
+// WalkerConfig field doc. Returns nil when the extension is absent.
+// Malformed values warn and return nil rather than failing the whole parse.
+func readWalkerExtension(extensions map[string]any, context string) *spec.WalkerConfig {
+ if extensions == nil {
+ return nil
+ }
+ raw, ok := extensions[extensionSyncWalker]
+ if !ok {
+ return nil
+ }
+ data, err := json.Marshal(raw)
+ if err != nil {
+ warnf("%s: %s: marshaling: %v; ignoring", context, extensionSyncWalker, err)
+ return nil
+ }
+ var cfg spec.WalkerConfig
+ if err := json.Unmarshal(data, &cfg); err != nil {
+ warnf("%s: %s: parsing: %v; ignoring", context, extensionSyncWalker, err)
+ return nil
+ }
+ if strings.TrimSpace(cfg.Parent) == "" {
+ warnf("%s: %s: parent is required; ignoring", context, extensionSyncWalker)
+ return nil
+ }
+ return &cfg
+}
+
// resolveIDFieldFromResponseSchema implements tiers 2-5 of the IDField fallback
// chain: prefer "id", then a resource-prefixed key (`<singular>_id` /
// `_uuid` / `_guid`), then "name", then the first scalar field listed in the
diff --git a/internal/openapi/parser_test.go b/internal/openapi/parser_test.go
index 52b44bb5..a39d74ab 100644
--- a/internal/openapi/parser_test.go
+++ b/internal/openapi/parser_test.go
@@ -5038,3 +5038,77 @@ func findParsedEndpointByPath(t *testing.T, parsed *spec.APISpec, method, path s
t.Fatalf("endpoint %s %s not found", method, path)
return spec.Endpoint{}
}
+
+// TestParseSyncWalkerExtension pins the x-pp-sync-walker operation
+// extension shape. The extension declares a hierarchical-walk dependency
+// for a child endpoint (parent resource name, optional non-PK key field,
+// optional explicit key param). Parsed into Endpoint.Walker.
+func TestParseSyncWalkerExtension(t *testing.T) {
+ t.Parallel()
+ data := []byte(`
+openapi: 3.0.3
+info:
+ title: Walker API
+ version: 1.0.0
+servers:
+ - url: https://api.example.com
+paths:
+ /games:
+ get:
+ summary: List games
+ responses:
+ "200":
+ description: ok
+ /games/{game_key}/leagues:
+ get:
+ summary: List leagues for a game
+ x-pp-sync-walker:
+ parent: games
+ key_field: game_key
+ key_param: game_key
+ parameters:
+ - name: game_key
+ in: path
+ required: true
+ schema:
+ type: string
+ responses:
+ "200":
+ description: ok
+ /leagues/{league_id}/teams:
+ get:
+ summary: List teams (walker without key_field)
+ x-pp-sync-walker:
+ parent: leagues
+ parameters:
+ - name: league_id
+ in: path
+ required: true
+ schema:
+ type: string
+ responses:
+ "200":
+ description: ok
+`)
+
+ parsed, err := Parse(data)
+ require.NoError(t, err)
+
+ // Endpoint with full walker config.
+ leagues := findParsedEndpointByPath(t, parsed, "GET", "/games/{game_key}/leagues")
+ require.NotNil(t, leagues.Walker, "x-pp-sync-walker must populate Endpoint.Walker")
+ assert.Equal(t, "games", leagues.Walker.Parent)
+ assert.Equal(t, "game_key", leagues.Walker.KeyField)
+ assert.Equal(t, "game_key", leagues.Walker.KeyParam)
+
+ // Endpoint with only parent set.
+ teams := findParsedEndpointByPath(t, parsed, "GET", "/leagues/{league_id}/teams")
+ require.NotNil(t, teams.Walker)
+ assert.Equal(t, "leagues", teams.Walker.Parent)
+ assert.Empty(t, teams.Walker.KeyField)
+ assert.Empty(t, teams.Walker.KeyParam)
+
+ // Endpoint without the extension.
+ games := findParsedEndpointByPath(t, parsed, "GET", "/games")
+ assert.Nil(t, games.Walker, "endpoint without x-pp-sync-walker must have nil Walker")
+}
diff --git a/internal/profiler/profiler.go b/internal/profiler/profiler.go
index b8c102cf..bed949a0 100644
--- a/internal/profiler/profiler.go
+++ b/internal/profiler/profiler.go
@@ -1,6 +1,8 @@
package profiler
import (
+ "fmt"
+ "os"
"slices"
"sort"
"strings"
@@ -129,6 +131,14 @@ type DependentResource struct {
// Discriminator routes heterogeneous dependent-resource response items to
// concrete typed resources before storage.
Discriminator DiscriminatorDispatch
+
+ // KeyField, when non-empty, names the field to extract from each parent
+ // record for substitution into the child path — overriding the default of
+ // using the parent's primary key (IDField on the parent's SyncableResource
+ // entry). Populated from a spec-declared walker (Endpoint.Walker.KeyField
+ // in internal YAML, or the `key_field` key under `x-pp-sync-walker` in
+ // OpenAPI). When empty, the existing parent-primary-key flow runs.
+ KeyField string
}
// APIProfile describes the shape of an API and what power-user features it warrants.
@@ -464,6 +474,7 @@ func Profile(s *spec.APISpec) *APIProfile {
p.SyncableResources = sortedSyncableResources(syncable)
p.DependentSyncResources = detectDependentResources(parameterized, syncable, shardedSubResources)
+ p.DependentSyncResources = applySpecWalkers(s, p.DependentSyncResources, syncable, s.Types, resourceNameIndex)
for resource, fields := range searchable {
p.SearchableFields[resource] = sortedKeys(fields)
}
@@ -1000,6 +1011,137 @@ func detectDependentResources(parameterized map[string]parameterizedEntry, synca
return deps
}
+// applySpecWalkers merges spec-declared walker configs (Endpoint.Walker,
+// populated from the `walker:` internal-YAML field or the `x-pp-sync-walker`
+// OpenAPI operation extension) into the dependent-sync set. For each endpoint
+// with a non-nil walker, the function either augments the matching
+// auto-detected DependentResource (carrying ParentResource, ParentIDParam,
+// and KeyField overrides through) or synthesizes a new entry when
+// auto-detection missed the link — covering paths where the placeholder name
+// does not match a parent resource, or paths with the placeholder in a
+// matrix or query parameter that resolveParentResource cannot map.
+//
+// Walker configs that fail validation are dropped with a stderr warning
+// rather than silently. Three checks fail:
+//
+// - parent is not a syncable resource: typo or stale spec; without a flat-
+// list parent endpoint there is nothing to iterate.
+// - the child path has 2+ {placeholders} and key_param is not declared
+// explicitly: firstPathParam returns the first placeholder, which on a
+// 2-deep path is the parent slot, almost always wrong.
+// - the child path has 0 placeholders and key_param is not declared (the
+// walker would bind via matrix/query but has no slot named).
+//
+// Existing dependent entries are matched by ("GET "+path) tuple — walker is
+// sync-only and GET-only, and a path-only key would collide if two endpoints
+// share a path across resources or methods.
+//
+// Synthesized entries derive Name from spec.ToSnakeCase(resourceName), not
+// from the endpoint-map key, so a walker that re-declares an already-auto-
+// detected path doesn't create a parallel entry under a different Name.
+// All other per-endpoint fields (Tier, IDField, Critical, SinceParam,
+// Discriminator) flow through metaFromEndpoint so the synthesized entry
+// matches what detectDependentResources would have produced — incremental
+// sync, tier routing, and discriminator dispatch all work the same.
+//
+// Entries without a walker pass through unchanged.
+func applySpecWalkers(s *spec.APISpec, deps []DependentResource, syncable map[string]syncableMeta, types map[string]spec.TypeDef, resourceNameIndex map[string]string) []DependentResource {
+ if s == nil {
+ return deps
+ }
+ byPath := make(map[string]int, len(deps))
+ for i, d := range deps {
+ byPath["GET "+d.Path] = i
+ }
+ var walk func(name string, r spec.Resource)
+ walk = func(resourceName string, r spec.Resource) {
+ for endpointName, e := range r.Endpoints {
+ if e.Walker == nil {
+ continue
+ }
+ parent := strings.ToLower(strings.TrimSpace(e.Walker.Parent))
+ if _, ok := syncable[parent]; !ok {
+ fmt.Fprintf(os.Stderr,
+ "warning: walker on %s.%s: parent %q is not a syncable resource; ignoring\n",
+ resourceName, endpointName, e.Walker.Parent)
+ continue
+ }
+ keyParam := strings.TrimSpace(e.Walker.KeyParam)
+ if keyParam == "" {
+ placeholders := countPathPlaceholders(e.Path)
+ switch placeholders {
+ case 1:
+ if p, ok := firstPathParam(e.Path); ok {
+ keyParam = p
+ }
+ case 0:
+ fmt.Fprintf(os.Stderr,
+ "warning: walker on %s.%s: path %q has no {placeholder}; declare key_param explicitly\n",
+ resourceName, endpointName, e.Path)
+ continue
+ default:
+ fmt.Fprintf(os.Stderr,
+ "warning: walker on %s.%s: path %q has %d placeholders; declare key_param explicitly\n",
+ resourceName, endpointName, e.Path, placeholders)
+ continue
+ }
+ }
+ keyField := strings.TrimSpace(e.Walker.KeyField)
+ lookupKey := "GET " + e.Path
+ if idx, ok := byPath[lookupKey]; ok {
+ deps[idx].ParentResource = parent
+ if keyParam != "" {
+ deps[idx].ParentIDParam = keyParam
+ }
+ deps[idx].KeyField = keyField
+ continue
+ }
+ meta := metaFromEndpoint(s, r, e, types, resourceNameIndex)
+ deps = append(deps, DependentResource{
+ Name: spec.ToSnakeCase(resourceName),
+ ParentResource: parent,
+ ParentIDParam: keyParam,
+ Path: e.Path,
+ Tier: meta.Tier,
+ IDField: meta.IDField,
+ Critical: meta.Critical,
+ SinceParam: meta.SinceParam,
+ Discriminator: meta.Discriminator,
+ KeyField: keyField,
+ })
+ byPath[lookupKey] = len(deps) - 1
+ }
+ for subName, sub := range r.SubResources {
+ walk(subName, sub)
+ }
+ }
+ for name, r := range s.Resources {
+ walk(name, r)
+ }
+ sort.Slice(deps, func(i, j int) bool { return deps[i].Name < deps[j].Name })
+ return deps
+}
+
+// countPathPlaceholders counts the number of `{name}` substitution slots in
+// a path template. Used by applySpecWalkers to decide whether
+// firstPathParam's default is safe (single-placeholder path) or ambiguous
+// (zero or 2+).
+func countPathPlaceholders(path string) int {
+ n := 0
+ for i := 0; i < len(path); i++ {
+ if path[i] != '{' {
+ continue
+ }
+ j := strings.IndexByte(path[i:], '}')
+ if j < 0 {
+ break
+ }
+ n++
+ i += j
+ }
+ return n
+}
+
// firstPathParam returns the name of the first {param} in a path template.
func firstPathParam(path string) (string, bool) {
start := strings.Index(path, "{")
diff --git a/internal/profiler/profiler_test.go b/internal/profiler/profiler_test.go
index 5d290e7b..6e350b02 100644
--- a/internal/profiler/profiler_test.go
+++ b/internal/profiler/profiler_test.go
@@ -1,6 +1,8 @@
package profiler
import (
+ "bytes"
+ "os"
"testing"
"github.com/mvanhorn/cli-printing-press/v4/internal/spec"
@@ -8,6 +10,24 @@ import (
"github.com/stretchr/testify/require"
)
+// captureStderr swaps os.Stderr for a pipe, runs fn, and returns whatever
+// fn wrote to stderr. The swap is single-threaded — safe for go test's
+// per-package sequential execution; do not use across parallel subtests
+// that both touch stderr.
+func captureStderr(t *testing.T, fn func()) string {
+ t.Helper()
+ orig := os.Stderr
+ r, w, err := os.Pipe()
+ require.NoError(t, err)
+ os.Stderr = w
+ t.Cleanup(func() { os.Stderr = orig })
+ fn()
+ require.NoError(t, w.Close())
+ var buf bytes.Buffer
+ _, _ = buf.ReadFrom(r)
+ return buf.String()
+}
+
func TestProfilePetstore(t *testing.T) {
profile := Profile(petstoreSpec())
@@ -1420,3 +1440,259 @@ func TestProfileSyncableResourceShorterPathWinsMetadata(t *testing.T) {
assert.Equal(t, "winner", profile.SyncableResources[0].IDField)
assert.True(t, profile.SyncableResources[0].Critical)
}
+
+// TestProfileSpecWalker_AugmentsAutoDetected verifies that a spec-declared
+// walker on an already-auto-detected dependent endpoint overrides
+// ParentResource, ParentIDParam, and KeyField in place rather than creating
+// a duplicate entry. /orders/{account_id} would auto-detect "account_id" →
+// "accounts" (after _id stripping) — the walker redirects to "customers"
+// and pins a non-PK key.
+func TestProfileSpecWalker_AugmentsAutoDetected(t *testing.T) {
+ s := &spec.APISpec{
+ Name: "shop",
+ Resources: map[string]spec.Resource{
+ "accounts": {
+ Endpoints: map[string]spec.Endpoint{
+ "list": {Method: "GET", Path: "/accounts", Response: spec.ResponseDef{Type: "array"}},
+ },
+ },
+ "customers": {
+ Endpoints: map[string]spec.Endpoint{
+ "list": {Method: "GET", Path: "/customers", Response: spec.ResponseDef{Type: "array"}, IDField: "customer_key"},
+ },
+ },
+ "orders": {
+ Endpoints: map[string]spec.Endpoint{
+ "list": {
+ Method: "GET",
+ Path: "/accounts/{account_id}/orders",
+ Response: spec.ResponseDef{Type: "array"},
+ Walker: &spec.WalkerConfig{
+ Parent: "customers",
+ KeyField: "customer_key",
+ KeyParam: "account_id",
+ },
+ },
+ },
+ },
+ },
+ }
+
+ profile := Profile(s)
+ require.Len(t, profile.DependentSyncResources, 1, "augment must not duplicate the entry")
+ dep := profile.DependentSyncResources[0]
+ assert.Equal(t, "customers", dep.ParentResource, "walker must redirect parent away from auto-detect")
+ assert.Equal(t, "account_id", dep.ParentIDParam)
+ assert.Equal(t, "customer_key", dep.KeyField)
+ assert.Equal(t, "/accounts/{account_id}/orders", dep.Path)
+}
+
+// TestProfileSpecWalker_SynthesizesMissingDependent verifies that a spec-
+// declared walker creates a new DependentResource entry when auto-detection
+// would not have linked the endpoint, and that the synthesized Name comes
+// from the containing resource (matching detectDependentResources's naming
+// convention) rather than the endpoint map key.
+func TestProfileSpecWalker_SynthesizesMissingDependent(t *testing.T) {
+ s := &spec.APISpec{
+ Name: "fantasy",
+ Resources: map[string]spec.Resource{
+ "games": {
+ Endpoints: map[string]spec.Endpoint{
+ "list": {Method: "GET", Path: "/games", Response: spec.ResponseDef{Type: "array"}, IDField: "game_key"},
+ },
+ },
+ "leagues": {
+ Endpoints: map[string]spec.Endpoint{
+ "fetch_for_game": {
+ Method: "GET",
+ Path: "/games/{game_key}/leagues",
+ Response: spec.ResponseDef{Type: "array"},
+ Walker: &spec.WalkerConfig{
+ Parent: "games",
+ KeyField: "game_key",
+ },
+ },
+ },
+ },
+ },
+ }
+
+ profile := Profile(s)
+ // Exactly one dependent for the leagues endpoint, named from the
+ // containing resource ("leagues"), not the endpoint key
+ // ("fetch_for_game" → "fetch_for_game" via ToSnakeCase).
+ require.Len(t, profile.DependentSyncResources, 1)
+ dep := profile.DependentSyncResources[0]
+ assert.Equal(t, "leagues", dep.Name, "Name must come from resource, not endpoint key")
+ assert.Equal(t, "games", dep.ParentResource)
+ assert.Equal(t, "game_key", dep.KeyField)
+ assert.Equal(t, "game_key", dep.ParentIDParam, "single-placeholder path: KeyParam defaults to firstPathParam")
+ assert.Equal(t, "/games/{game_key}/leagues", dep.Path)
+}
+
+// TestProfileSpecWalker_SynthesizePropagatesSinceParam verifies that a
+// walker-synthesized DependentResource carries through endpoint-level
+// SinceParam — incremental sync stays available for walker-declared
+// hierarchical children, matching the auto-detect path's behavior.
+// Greptile flagged a P1 regression on the initial draft where the
+// synthesize branch dropped SinceParam (and Discriminator); this test
+// pins the fix.
+func TestProfileSpecWalker_SynthesizePropagatesSinceParam(t *testing.T) {
+ s := &spec.APISpec{
+ Name: "fantasy",
+ Resources: map[string]spec.Resource{
+ "games": {
+ Endpoints: map[string]spec.Endpoint{
+ "list": {Method: "GET", Path: "/games", Response: spec.ResponseDef{Type: "array"}, IDField: "game_key"},
+ },
+ },
+ "leagues": {
+ Endpoints: map[string]spec.Endpoint{
+ "list": {
+ Method: "GET",
+ Path: "/games/{game_key}/leagues",
+ Response: spec.ResponseDef{Type: "array"},
+ Params: []spec.Param{
+ {Name: "game_key", PathParam: true},
+ {Name: "since"},
+ },
+ Walker: &spec.WalkerConfig{
+ Parent: "games",
+ KeyField: "game_key",
+ },
+ },
+ },
+ },
+ },
+ }
+
+ profile := Profile(s)
+ require.Len(t, profile.DependentSyncResources, 1)
+ dep := profile.DependentSyncResources[0]
+ assert.Equal(t, "leagues", dep.Name)
+ assert.Equal(t, "since", dep.SinceParam,
+ "synthesize branch must propagate SinceParam via metaFromEndpoint — incremental sync depends on it")
+}
+
+// TestProfileSpecWalker_NonSyncableParentWarns verifies that a walker
+// pointing at a non-syncable parent emits a stderr warning and is dropped.
+// Explicit walker:: declarations carry author intent; silently dropping a
+// typo'd parent would produce passing builds with missing data.
+func TestProfileSpecWalker_NonSyncableParentWarns(t *testing.T) {
+ s := &spec.APISpec{
+ Name: "fantasy",
+ Resources: map[string]spec.Resource{
+ // "sports" is not syncable (GET-by-id only, no list).
+ "sports": {
+ Endpoints: map[string]spec.Endpoint{
+ "get": {Method: "GET", Path: "/sports/{sport_id}", Response: spec.ResponseDef{Type: "object"}},
+ },
+ },
+ "leagues": {
+ Endpoints: map[string]spec.Endpoint{
+ "list": {
+ Method: "GET",
+ Path: "/leagues",
+ Response: spec.ResponseDef{Type: "array"},
+ Walker: &spec.WalkerConfig{
+ Parent: "sports",
+ KeyField: "sport_key",
+ },
+ },
+ },
+ },
+ },
+ }
+
+ var profile *APIProfile
+ stderr := captureStderr(t, func() {
+ profile = Profile(s)
+ })
+
+ assert.Contains(t, stderr, "warning: walker on leagues.list")
+ assert.Contains(t, stderr, `parent "sports" is not a syncable resource`)
+ for _, dep := range profile.DependentSyncResources {
+ assert.NotEqual(t, "leagues", dep.Name,
+ "walker with non-syncable parent must be dropped, not produce a DependentResource")
+ }
+}
+
+// TestProfileSpecWalker_MultiPlaceholderPathWarns verifies that a walker on
+// a path with 2+ {...} placeholders requires an explicit key_param. Without
+// it, firstPathParam's "first wins" default would silently pick the parent
+// slot on a 2-deep path — almost always the wrong slot for the child.
+// With explicit key_param, the walker is accepted.
+func TestProfileSpecWalker_MultiPlaceholderPathWarns(t *testing.T) {
+ t.Run("ambiguous: warn and drop", func(t *testing.T) {
+ s := &spec.APISpec{
+ Name: "fantasy",
+ Resources: map[string]spec.Resource{
+ "games": {
+ Endpoints: map[string]spec.Endpoint{
+ "list": {Method: "GET", Path: "/games", Response: spec.ResponseDef{Type: "array"}, IDField: "game_key"},
+ },
+ },
+ "rosters": {
+ Endpoints: map[string]spec.Endpoint{
+ "list": {
+ Method: "GET",
+ Path: "/games/{game_key}/leagues/{league_id}/roster",
+ Response: spec.ResponseDef{Type: "array"},
+ Walker: &spec.WalkerConfig{
+ Parent: "games",
+ // no key_param — ambiguous on 2-placeholder path
+ },
+ },
+ },
+ },
+ },
+ }
+ var profile *APIProfile
+ stderr := captureStderr(t, func() {
+ profile = Profile(s)
+ })
+ assert.Contains(t, stderr, "warning: walker on rosters.list")
+ assert.Contains(t, stderr, "2 placeholders")
+ assert.Contains(t, stderr, "declare key_param explicitly")
+ for _, dep := range profile.DependentSyncResources {
+ assert.NotEqual(t, "rosters", dep.Name, "ambiguous walker must be dropped")
+ }
+ })
+
+ t.Run("explicit key_param: accepted", func(t *testing.T) {
+ s := &spec.APISpec{
+ Name: "fantasy",
+ Resources: map[string]spec.Resource{
+ "games": {
+ Endpoints: map[string]spec.Endpoint{
+ "list": {Method: "GET", Path: "/games", Response: spec.ResponseDef{Type: "array"}, IDField: "game_key"},
+ },
+ },
+ "rosters": {
+ Endpoints: map[string]spec.Endpoint{
+ "list": {
+ Method: "GET",
+ Path: "/games/{game_key}/leagues/{league_id}/roster",
+ Response: spec.ResponseDef{Type: "array"},
+ Walker: &spec.WalkerConfig{
+ Parent: "games",
+ KeyField: "game_key",
+ KeyParam: "league_id",
+ },
+ },
+ },
+ },
+ },
+ }
+ profile := Profile(s)
+ var found bool
+ for _, dep := range profile.DependentSyncResources {
+ if dep.Name == "rosters" {
+ found = true
+ assert.Equal(t, "league_id", dep.ParentIDParam, "explicit key_param must be used verbatim")
+ assert.Equal(t, "game_key", dep.KeyField)
+ }
+ }
+ assert.True(t, found, "walker with explicit key_param must produce a dependent entry")
+ })
+}
diff --git a/internal/spec/spec.go b/internal/spec/spec.go
index b23c0397..6aa6f2f8 100644
--- a/internal/spec/spec.go
+++ b/internal/spec/spec.go
@@ -992,8 +992,37 @@ type Endpoint struct {
// true, a per-resource failure is treated as a hard failure even under the
// new (non-strict) exit-code policy. Populated from the path-item-level
// `x-critical` extension on OpenAPI specs; defaults to false.
- Critical bool `yaml:"critical,omitempty" json:"critical,omitempty"`
- Alias string `yaml:"-" json:"-"` // computed, not from YAML
+ Critical bool `yaml:"critical,omitempty" json:"critical,omitempty"`
+ // Walker, when present, declares this endpoint as a hierarchical child
+ // resource fetched by iterating a named parent. Used when the generator's
+ // path-param dependent-resource auto-detection would miss the link — for
+ // example when the child's path puts the parent placeholder in a matrix
+ // or query parameter, or when the placeholder name does not match the
+ // parent resource. Internal YAML emits it as `walker:` on the endpoint;
+ // OpenAPI emits it as `x-pp-sync-walker` on the operation. See
+ // docs/SPEC-EXTENSIONS.md for the canonical schema.
+ Walker *WalkerConfig `yaml:"walker,omitempty" json:"walker,omitempty"`
+ Alias string `yaml:"-" json:"-"` // computed, not from YAML
+}
+
+// WalkerConfig declares a hierarchical-walk dependency for a child endpoint.
+// The generator synthesizes (or augments) a DependentResource entry from this
+// config so the existing dependent-sync machinery handles the fan-out.
+type WalkerConfig struct {
+ // Parent is the resource name to iterate. Must be syncable (i.e., have a
+ // flat-list endpoint) so its rows are available in the local store.
+ Parent string `yaml:"parent" json:"parent"`
+ // KeyField is the field name to extract from each parent record for
+ // substitution into the child path. Defaults to the parent's IDField
+ // (primary key) when empty. Use this when the child path needs a parent
+ // field that is not the parent's primary key.
+ KeyField string `yaml:"key_field,omitempty" json:"key_field,omitempty"`
+ // KeyParam is the placeholder name in the child path that receives the
+ // extracted key value. Defaults to the first {placeholder} found in the
+ // child's Path when empty. Set this explicitly when the child path has
+ // multiple placeholders or when the placeholder name does not match the
+ // auto-detection convention.
+ KeyParam string `yaml:"key_param,omitempty" json:"key_param,omitempty"`
}
func (e Endpoint) EffectiveResponseFormat() string {
diff --git a/internal/spec/spec_test.go b/internal/spec/spec_test.go
index 970d0f1f..6b1f46fb 100644
--- a/internal/spec/spec_test.go
+++ b/internal/spec/spec_test.go
@@ -3760,3 +3760,58 @@ func TestValidateRejectsReservedPlaceholderHost(t *testing.T) {
})
}
}
+
+// TestWalkerConfig_YAMLRoundTrip catches future regressions in WalkerConfig
+// YAML tags or the Walker field's omitempty on Endpoint. The Walker pointer
+// and all three sub-fields must survive a marshal → unmarshal cycle.
+func TestWalkerConfig_YAMLRoundTrip(t *testing.T) {
+ t.Parallel()
+
+ t.Run("populated walker survives round-trip", func(t *testing.T) {
+ t.Parallel()
+ ep := Endpoint{
+ Method: "GET",
+ Path: "/games/{game_key}/leagues",
+ Walker: &WalkerConfig{
+ Parent: "games",
+ KeyField: "game_key",
+ KeyParam: "game_key",
+ },
+ }
+ data, err := yaml.Marshal(ep)
+ require.NoError(t, err)
+ assert.Contains(t, string(data), "walker:")
+ assert.Contains(t, string(data), "parent: games")
+ assert.Contains(t, string(data), "key_field: game_key")
+ assert.Contains(t, string(data), "key_param: game_key")
+
+ var roundTripped Endpoint
+ require.NoError(t, yaml.Unmarshal(data, &roundTripped))
+ require.NotNil(t, roundTripped.Walker)
+ assert.Equal(t, "games", roundTripped.Walker.Parent)
+ assert.Equal(t, "game_key", roundTripped.Walker.KeyField)
+ assert.Equal(t, "game_key", roundTripped.Walker.KeyParam)
+ })
+
+ t.Run("nil walker omits the section", func(t *testing.T) {
+ t.Parallel()
+ ep := Endpoint{Method: "GET", Path: "/games"}
+ data, err := yaml.Marshal(ep)
+ require.NoError(t, err)
+ assert.NotContains(t, string(data), "walker:")
+ })
+
+ t.Run("walker with only parent omits optional sub-fields", func(t *testing.T) {
+ t.Parallel()
+ ep := Endpoint{
+ Method: "GET",
+ Path: "/leagues/{league_id}/teams",
+ Walker: &WalkerConfig{Parent: "leagues"},
+ }
+ data, err := yaml.Marshal(ep)
+ require.NoError(t, err)
+ assert.Contains(t, string(data), "parent: leagues")
+ assert.NotContains(t, string(data), "key_field")
+ assert.NotContains(t, string(data), "key_param")
+ })
+}
diff --git a/testdata/golden/cases/generate-sync-walker-api/artifacts.txt b/testdata/golden/cases/generate-sync-walker-api/artifacts.txt
new file mode 100644
index 00000000..cd48ab85
--- /dev/null
+++ b/testdata/golden/cases/generate-sync-walker-api/artifacts.txt
@@ -0,0 +1,5 @@
+sync-walker-golden/.printing-press.json
+sync-walker-golden/internal/cli/sync.go
+sync-walker-golden/internal/cli/promoted_games.go
+sync-walker-golden/internal/cli/promoted_leagues.go
+sync-walker-golden/internal/store/store.go
diff --git a/testdata/golden/cases/generate-sync-walker-api/command.txt b/testdata/golden/cases/generate-sync-walker-api/command.txt
new file mode 100644
index 00000000..0921c44b
--- /dev/null
+++ b/testdata/golden/cases/generate-sync-walker-api/command.txt
@@ -0,0 +1,2 @@
+set -euo pipefail
+"$BINARY" generate --spec testdata/golden/fixtures/sync-walker-api.yaml --output "$CASE_ACTUAL_DIR/sync-walker-golden" --force --spec-url file://testdata/golden/fixtures/sync-walker-api.yaml --validate=false
diff --git a/testdata/golden/expected/generate-golden-api/printing-press-golden/internal/cli/sync.go b/testdata/golden/expected/generate-golden-api/printing-press-golden/internal/cli/sync.go
index 35d48144..d77225d6 100644
--- a/testdata/golden/expected/generate-golden-api/printing-press-golden/internal/cli/sync.go
+++ b/testdata/golden/expected/generate-golden-api/printing-press-golden/internal/cli/sync.go
@@ -943,16 +943,22 @@ func syncResourcePath(resource string) (string, error) {
}
// dependentResourceDef describes a child resource that requires iterating parent IDs to sync.
+// When KeyField is non-empty, the dependent's parent IDs are extracted from the parent
+// records' KeyField via json_extract rather than from the parent table's primary key.
+// Populated from a spec-declared walker (Endpoint.Walker.KeyField in internal YAML,
+// or `key_field` under `x-pp-sync-walker` in OpenAPI). Empty KeyField preserves the
+// existing parent-primary-key flow byte-for-byte.
type dependentResourceDef struct {
Name string
ParentTable string
ParentIDParam string
PathTemplate string
+ KeyField string
}
func dependentResourceDefs() []dependentResourceDef {
return []dependentResourceDef{
- {Name: "tasks", ParentTable: "projects", ParentIDParam: "projectId", PathTemplate: "/projects/{projectId}/tasks"},
+ {Name: "tasks", ParentTable: "projects", ParentIDParam: "projectId", PathTemplate: "/projects/{projectId}/tasks", KeyField: ""},
}
}
@@ -985,8 +991,19 @@ func syncDependentResource(c interface {
}, db *store.Store, dep dependentResourceDef, sinceTS string, full bool, maxPages int) syncResult {
started := time.Now()
- // Query parent table for all IDs
- parentIDs, err := db.ListIDs(dep.ParentTable)
+ // Query parent table for the keys to substitute into the child path.
+ // When KeyField is empty, use the parent's primary key via ListIDs
+ // (the original flat parent-child flow). When KeyField is set, the
+ // spec declared a walker that extracts a non-PK field from each parent
+ // record — ListField looks up the field in the parent's typed column
+ // if present, otherwise json_extract from the generic resources table.
+ var parentIDs []string
+ var err error
+ if dep.KeyField != "" {
+ parentIDs, err = db.ListField(dep.ParentTable, dep.KeyField)
+ } else {
+ parentIDs, err = db.ListIDs(dep.ParentTable)
+ }
if err != nil || len(parentIDs) == 0 {
if len(parentIDs) == 0 {
if humanFriendly {
diff --git a/testdata/golden/expected/generate-sync-walker-api/exit.txt b/testdata/golden/expected/generate-sync-walker-api/exit.txt
new file mode 100644
index 00000000..573541ac
--- /dev/null
+++ b/testdata/golden/expected/generate-sync-walker-api/exit.txt
@@ -0,0 +1 @@
+0
diff --git a/testdata/golden/expected/generate-sync-walker-api/stderr.txt b/testdata/golden/expected/generate-sync-walker-api/stderr.txt
new file mode 100644
index 00000000..6734839d
--- /dev/null
+++ b/testdata/golden/expected/generate-sync-walker-api/stderr.txt
@@ -0,0 +1,2 @@
+warning: could not derive run_id from --research-dir; phase5 dogfood acceptance will refuse to write without it
+Generated sync-walker-golden at <ARTIFACT_DIR>/generate-sync-walker-api/sync-walker-golden
diff --git a/testdata/golden/expected/generate-sync-walker-api/stdout.txt b/testdata/golden/expected/generate-sync-walker-api/stdout.txt
new file mode 100644
index 00000000..e69de29b
diff --git a/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/.printing-press.json b/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/.printing-press.json
new file mode 100644
index 00000000..1d7d8a71
--- /dev/null
+++ b/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/.printing-press.json
@@ -0,0 +1,23 @@
+{
+ "api_name": "sync-walker-golden",
+ "api_version": "1.0.0",
+ "auth_env_vars": [
+ "WALKER_GOLDEN_TOKEN"
+ ],
+ "auth_type": "bearer_token",
+ "cli_name": "sync-walker-golden-pp-cli",
+ "display_name": "Sync Walker Golden",
+ "generated_at": "<GENERATED_AT>",
+ "mcp_binary": "sync-walker-golden-pp-mcp",
+ "mcp_ready": "full",
+ "mcp_tool_count": 2,
+ "owner": "printing-press-golden",
+ "printer": "printing-press-golden",
+ "printer_name": "printing-press-golden",
+ "printing_press_version": "<PRINTING_PRESS_VERSION>",
+ "schema_version": 1,
+ "spec_checksum": "sha256:2aa39c064cd068cb090102e22a446f9542fa06f75453d52e4fa9db5c21f27809",
+ "spec_format": "internal",
+ "spec_path": "testdata/golden/fixtures/sync-walker-api.yaml",
+ "spec_url": "file://testdata/golden/fixtures/sync-walker-api.yaml"
+}
diff --git a/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/cli/promoted_games.go b/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/cli/promoted_games.go
new file mode 100644
index 00000000..219562c3
--- /dev/null
+++ b/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/cli/promoted_games.go
@@ -0,0 +1,84 @@
+// Copyright 2026 printing-press-golden. Licensed under Apache-2.0. See LICENSE.
+// Generated by CLI Printing Press (https://github.com/mvanhorn/cli-printing-press). DO NOT EDIT.
+
+package cli
+
+import (
+ "encoding/json"
+ "fmt"
+ "os"
+
+ "github.com/spf13/cobra"
+)
+
+func newGamesPromotedCmd(flags *rootFlags) *cobra.Command {
+
+ cmd := &cobra.Command{
+ Use: "games",
+ Short: "List games",
+ Long: "Shortcut for 'games list'. List games",
+ Example: " sync-walker-golden-pp-cli games",
+ Annotations: map[string]string{"pp:endpoint": "games.list", "pp:method": "GET", "pp:path": "/games", "mcp:read-only": "true"},
+ RunE: func(cmd *cobra.Command, args []string) error {
+ c, err := flags.newClient()
+ if err != nil {
+ return err
+ }
+
+ path := "/games"
+ params := map[string]string{}
+ data, prov, err := resolveRead(cmd.Context(), c, flags, "games", false, path, params, nil)
+ if err != nil {
+ return classifyAPIError(err, flags)
+ }
+ // Unwrap API response envelopes (e.g. {"status":"success","data":[...]})
+ // so output helpers see the inner data, not the wrapper.
+ data = extractResponseData(data)
+
+ // Print provenance to stderr
+ {
+ var countItems []json.RawMessage
+ if json.Unmarshal(data, &countItems) != nil {
+ // Single object, not an array
+ countItems = []json.RawMessage{data}
+ }
+ printProvenance(cmd, len(countItems), prov)
+ }
+ // For JSON output, wrap with provenance envelope. --select wins over
+ // --compact when both are set; --compact only runs when no explicit
+ // fields were requested. Explicit format flags (--csv, --quiet, --plain)
+ // opt out of the auto-JSON path so piped consumers that asked for a
+ // non-JSON format reach the standard pipeline below.
+ if flags.asJSON || (!isTerminal(cmd.OutOrStdout()) && !flags.csv && !flags.quiet && !flags.plain) {
+ filtered := data
+ if flags.selectFields != "" {
+ filtered = filterFields(filtered, flags.selectFields)
+ } else if flags.compact {
+ filtered = compactFields(filtered)
+ }
+ wrapped, wrapErr := wrapWithProvenance(filtered, prov)
+ if wrapErr != nil {
+ return wrapErr
+ }
+ return printOutput(cmd.OutOrStdout(), wrapped, true)
+ }
+ if wantsHumanTable(cmd.OutOrStdout(), flags) {
+ var items []map[string]any
+ if json.Unmarshal(data, &items) == nil && len(items) > 0 {
+ if err := printAutoTable(cmd.OutOrStdout(), items); err != nil {
+ return err
+ }
+ if len(items) >= 25 {
+ fmt.Fprintf(os.Stderr, "\nShowing %d results. To narrow: add --limit, --json --select, or filter flags.\n", len(items))
+ }
+ return nil
+ }
+ }
+ return printOutputWithFlags(cmd.OutOrStdout(), data, flags)
+ },
+ }
+
+ // Wire sibling endpoints and sub-resources as subcommands
+
+ return cmd
+}
diff --git a/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/cli/promoted_leagues.go b/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/cli/promoted_leagues.go
new file mode 100644
index 00000000..31ae1d76
--- /dev/null
+++ b/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/cli/promoted_leagues.go
@@ -0,0 +1,98 @@
+// Copyright 2026 printing-press-golden. Licensed under Apache-2.0. See LICENSE.
+// Generated by CLI Printing Press (https://github.com/mvanhorn/cli-printing-press). DO NOT EDIT.
+
+package cli
+
+import (
+ "encoding/json"
+ "fmt"
+ "os"
+
+ "github.com/spf13/cobra"
+)
+
+func newLeaguesPromotedCmd(flags *rootFlags) *cobra.Command {
+
+ cmd := &cobra.Command{
+ Use: "leagues <game_key>",
+ Short: "List leagues for a game",
+ Long: "Shortcut for 'leagues list'. List leagues for a game",
+ Example: " sync-walker-golden-pp-cli leagues your-token-here",
+ Annotations: map[string]string{"pp:endpoint": "leagues.list", "pp:method": "GET", "pp:path": "/games/{game_key}/leagues", "mcp:read-only": "true"},
+ RunE: func(cmd *cobra.Command, args []string) error {
+ c, err := flags.newClient()
+ if err != nil {
+ return err
+ }
+
+ path := "/games/{game_key}/leagues"
+ if len(args) < 1 {
+ // JSON envelope: {error, usage}. Written first; the
+ // usageErr return preserves exit code 2 across modes.
+ if flags.asJSON {
+ if printErr := printJSONFiltered(cmd.OutOrStdout(), map[string]any{
+ "error": "game_key is required",
+ "usage": fmt.Sprintf("%s <%s>", cmd.CommandPath(), "game_key"),
+ }, flags); printErr != nil {
+ return printErr
+ }
+ }
+ return usageErr(fmt.Errorf("game_key is required\nUsage: %s <%s>", cmd.CommandPath(), "game_key"))
+ }
+ path = replacePathParam(path, "game_key", args[0])
+ params := map[string]string{}
+ data, prov, err := resolveRead(cmd.Context(), c, flags, "leagues", false, path, params, nil)
+ if err != nil {
+ return classifyAPIError(err, flags)
+ }
+ // Unwrap API response envelopes (e.g. {"status":"success","data":[...]})
+ // so output helpers see the inner data, not the wrapper.
+ data = extractResponseData(data)
+
+ // Print provenance to stderr
+ {
+ var countItems []json.RawMessage
+ if json.Unmarshal(data, &countItems) != nil {
+ // Single object, not an array
+ countItems = []json.RawMessage{data}
+ }
+ printProvenance(cmd, len(countItems), prov)
+ }
+ // For JSON output, wrap with provenance envelope. --select wins over
+ // --compact when both are set; --compact only runs when no explicit
+ // fields were requested. Explicit format flags (--csv, --quiet, --plain)
+ // opt out of the auto-JSON path so piped consumers that asked for a
+ // non-JSON format reach the standard pipeline below.
+ if flags.asJSON || (!isTerminal(cmd.OutOrStdout()) && !flags.csv && !flags.quiet && !flags.plain) {
+ filtered := data
+ if flags.selectFields != "" {
+ filtered = filterFields(filtered, flags.selectFields)
+ } else if flags.compact {
+ filtered = compactFields(filtered)
+ }
+ wrapped, wrapErr := wrapWithProvenance(filtered, prov)
+ if wrapErr != nil {
+ return wrapErr
+ }
+ return printOutput(cmd.OutOrStdout(), wrapped, true)
+ }
+ if wantsHumanTable(cmd.OutOrStdout(), flags) {
+ var items []map[string]any
+ if json.Unmarshal(data, &items) == nil && len(items) > 0 {
+ if err := printAutoTable(cmd.OutOrStdout(), items); err != nil {
+ return err
+ }
+ if len(items) >= 25 {
+ fmt.Fprintf(os.Stderr, "\nShowing %d results. To narrow: add --limit, --json --select, or filter flags.\n", len(items))
+ }
+ return nil
+ }
+ }
+ return printOutputWithFlags(cmd.OutOrStdout(), data, flags)
+ },
+ }
+
+ // Wire sibling endpoints and sub-resources as subcommands
+
+ return cmd
+}
diff --git a/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/cli/sync.go b/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/cli/sync.go
new file mode 100644
index 00000000..b722397c
--- /dev/null
+++ b/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/cli/sync.go
@@ -0,0 +1,1247 @@
+// Copyright 2026 printing-press-golden. Licensed under Apache-2.0. See LICENSE.
+// Generated by CLI Printing Press (https://github.com/mvanhorn/cli-printing-press). DO NOT EDIT.
+
+package cli
+
+import (
+ "encoding/json"
+ "fmt"
+ "net/url"
+ "os"
+ "regexp"
+ "strconv"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "time"
+ "sync-walker-golden-pp-cli/internal/store"
+ "github.com/spf13/cobra"
+)
+
+// unresolvedPathKeyRE matches `{key}` placeholders left in a sync path
+// after syncResourcePath() resolution. Hierarchical APIs (Yahoo Fantasy,
+// Reddit pre-2024, YouTube Data v3, MLB Stats, etc.) declare paths like
+// "/league/{league_key}/players" that can only be filled from parent
+// context — flat-list sync cannot fill them. Resources with unresolved
+// keys emit sync_warning and are skipped without aborting the run, so
+// sync still completes for resources that DO have resolvable paths.
+var unresolvedPathKeyRE = regexp.MustCompile(`\{[a-zA-Z_][a-zA-Z0-9_]*\}`)
+
+// syncResult holds the outcome of syncing a single resource.
+type syncResult struct {
+ Resource string
+ Count int
+ Err error
+ Warn error
+ Duration time.Duration
+}
+
+func newSyncCmd(flags *rootFlags) *cobra.Command {
+ var resources []string
+ var full bool
+ var since string
+ var concurrency int
+ var dbPath string
+ var maxPages int
+ var latestOnly bool
+ var strict bool
+
+ cmd := &cobra.Command{
+ Use: "sync",
+ Short: "Sync API data to local SQLite for offline search and analysis",
+ Long: `Sync data from the API into a local SQLite database. Supports resumable
+incremental sync (only fetches new data since last sync) and full resync.
+Once synced, use the 'search' command for instant full-text search.
+
+Exit codes & warnings:
+ Resources the API denies access to (HTTP 403, or HTTP 400 with an
+ access-policy body) are reported as warnings rather than failing the
+ run. In --json mode each is emitted as a {"event":"sync_warning",...}
+ line carrying status, reason, and message fields, and a final
+ {"event":"sync_summary",...} aggregates the run.
+
+ Exit 0 when at least one resource synced and no resource flagged in
+ the spec as critical (x-critical: true) failed; non-critical failures
+ emit {"event":"sync_warning","reason":"exit_policy_default_changed",
+ ...} so callers can detect that a partial failure was tolerated. Pass
+ --strict to exit non-zero on any per-resource failure. Exit is always
+ non-zero when every selected resource failed, regardless of --strict.`,
+ Example: ` # Sync all resources
+ sync-walker-golden-pp-cli sync
+
+ # Sync specific resources only
+ sync-walker-golden-pp-cli sync --resources channels,messages
+
+ # Full resync (ignore previous checkpoint)
+ sync-walker-golden-pp-cli sync --full
+
+ # Incremental sync: only records from the last 7 days
+ sync-walker-golden-pp-cli sync --since 7d
+
+ # Parallel sync with 8 workers
+ sync-walker-golden-pp-cli sync --concurrency 8
+
+ # Latest-only: refresh head of each resource, no historical backfill
+ sync-walker-golden-pp-cli sync --latest-only`,
+ RunE: func(cmd *cobra.Command, args []string) error {
+ c, err := flags.newClient()
+ if err != nil {
+ return err
+ }
+ c.NoCache = true
+
+ if dbPath == "" {
+ dbPath = defaultDBPath("sync-walker-golden-pp-cli")
+ }
+
+ db, err := store.OpenWithContext(cmd.Context(), dbPath)
+ if err != nil {
+ return fmt.Errorf("opening local database: %w", err)
+ }
+ defer db.Close()
+ // Snapshot before defaults expand, so an empty user filter stays empty
+ // and dependents inherit "sync everything" instead of the default list.
+ parentFilter := append([]string(nil), resources...)
+
+ // If no specific resources, sync top-level resources
+ if len(resources) == 0 {
+ resources = defaultSyncResources()
+ }
+
+ // --full: clear all sync cursors before starting
+ if full {
+ for _, resource := range resources {
+ _ = db.SaveSyncState(resource, "", 0)
+ }
+ }
+
+ // --latest-only narrows to the first page of each resource
+ // ignoring the historical resume cursor. We cap maxPages at 1
+ // here rather than re-interpreting it downstream so the rest
+ // of the sync loop stays oblivious. Mutually-useful with
+ // --since: if the user set --since, that threshold still wins
+ // and we don't short-circuit historical context they asked for.
+ if latestOnly {
+ if since == "" {
+ maxPages = 1
+ // Clear the cursor so we start from the head each time;
+ // the goal of --latest-only is "refresh the top" not
+ // "resume from wherever I left off".
+ for _, resource := range resources {
+ existing, _, _, _ := db.GetSyncState(resource)
+ if existing != "" {
+ _ = db.SaveSyncState(resource, "", 0)
+ }
+ }
+ } else if humanFriendly {
+ fmt.Fprintln(os.Stderr, "warning: --latest-only ignored because --since is set; --since takes precedence")
+ }
+ }
+
+ // Resolve --since into an RFC3339 timestamp
+ sinceTS := ""
+ if since != "" {
+ ts, err := parseSinceDuration(since)
+ if err != nil {
+ return fmt.Errorf("invalid --since value %q: %w", since, err)
+ }
+ sinceTS = ts.Format(time.RFC3339)
+ }
+
+ // Worker pool: produce resources, N workers consume
+ if concurrency < 1 {
+ concurrency = 4
+ }
+
+ started := time.Now()
+ work := make(chan string, len(resources))
+ results := make(chan syncResult, len(resources))
+
+ var wg sync.WaitGroup
+ for i := 0; i < concurrency; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for resource := range work {
+ res := syncResource(c, db, resource, sinceTS, full, maxPages)
+ results <- res
+ }
+ }()
+ }
+
+ // Enqueue all resources
+ for _, resource := range resources {
+ work <- resource
+ }
+ close(work)
+
+ // Collect results in a separate goroutine
+ go func() {
+ wg.Wait()
+ close(results)
+ }()
+
+ var totalSynced int
+ var errCount int
+ var criticalErrCount int
+ var warnCount int
+ var successCount int
+ for res := range results {
+ if res.Err != nil {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, " %s: error: %v\n", res.Resource, res.Err)
+ }
+ errCount++
+ if criticalResources[res.Resource] {
+ criticalErrCount++
+ }
+ } else if res.Warn != nil {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, " %s: warning: %v\n", res.Resource, res.Warn)
+ }
+ warnCount++
+ } else {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, " %s: %d synced (done)\n", res.Resource, res.Count)
+ }
+ totalSynced += res.Count
+ successCount++
+ }
+ }
+ // Sync dependent (parent-child) resources sequentially after flat resources.
+ depResults := syncDependentResources(c, db, sinceTS, full, maxPages, parentFilter)
+ for _, res := range depResults {
+ if res.Err != nil {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, " %s: error: %v\n", res.Resource, res.Err)
+ }
+ errCount++
+ if criticalResources[res.Resource] {
+ criticalErrCount++
+ }
+ } else if res.Warn != nil {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, " %s: warning: %v\n", res.Resource, res.Warn)
+ }
+ warnCount++
+ } else {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, " %s: %d synced (done)\n", res.Resource, res.Count)
+ }
+ totalSynced += res.Count
+ successCount++
+ }
+ }
+
+ elapsed := time.Since(started)
+ totalResources := successCount + warnCount + errCount
+ if humanFriendly {
+ if warnCount > 0 {
+ fmt.Fprintf(os.Stderr, "Sync complete: %d records across %d resources (%d warned, %.1fs)\n",
+ totalSynced, totalResources, warnCount, elapsed.Seconds())
+ } else {
+ fmt.Fprintf(os.Stderr, "Sync complete: %d records across %d resources (%.1fs)\n",
+ totalSynced, totalResources, elapsed.Seconds())
+ }
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_summary","total_records":%d,"resources":%d,"success":%d,"warned":%d,"errored":%d,"duration_ms":%d}`+"\n",
+ totalSynced, totalResources, successCount, warnCount, errCount, elapsed.Milliseconds())
+ }
+
+ // Exit-code policy:
+ // 1. --strict + any error -> non-zero (legacy contract)
+ // 2. any critical failure -> non-zero regardless of --strict
+ // 3. nothing synced -> non-zero (preserves "all-warned" / "all-errored" exit)
+ // 4. otherwise -> exit 0 (any data synced + no critical failed)
+ // When branch 4 suppresses what branch 1 would have rejected, emit a
+ // one-shot sync_warning with reason "exit_policy_default_changed" so
+ // CI scripts that depend on $? != 0 can discover the contract change
+ // without reading the CHANGELOG.
+ if strict && errCount > 0 {
+ return fmt.Errorf("%d resource(s) failed to sync", errCount)
+ }
+ if criticalErrCount > 0 {
+ return fmt.Errorf("%d critical resource(s) failed to sync", criticalErrCount)
+ }
+ if successCount == 0 {
+ if warnCount > 0 && errCount == 0 {
+ return fmt.Errorf("%d resource(s) skipped due to insufficient access", warnCount)
+ }
+ if errCount > 0 {
+ return fmt.Errorf("%d resource(s) failed to sync", errCount)
+ }
+ }
+ if errCount > 0 && !strict && criticalErrCount == 0 && successCount > 0 {
+ if !humanFriendly {
+ msg := fmt.Sprintf("%d resource(s) failed but exit code is 0 because the new default treats non-critical failures as warnings. Pass --strict to restore the old behavior, or annotate critical resources with x-critical: true. See CHANGELOG.", errCount)
+ fmt.Fprintf(os.Stdout, `{"event":"sync_warning","reason":"exit_policy_default_changed","errored":%d,"message":"%s"}`+"\n",
+ errCount, strings.ReplaceAll(msg, `"`, `\"`))
+ } else {
+ fmt.Fprintf(os.Stderr, "warning: %d resource(s) failed but exit code is 0 because the new default treats non-critical failures as warnings. Pass --strict to restore the old behavior, or annotate critical resources with x-critical: true.\n", errCount)
+ }
+ }
+ return nil
+ },
+ }
+
+ cmd.Flags().StringSliceVar(&resources, "resources", nil, "Comma-separated resource types to sync")
+ cmd.Flags().BoolVar(&full, "full", false, "Full resync (ignore previous checkpoint)")
+ cmd.Flags().StringVar(&since, "since", "", "Incremental sync duration (e.g. 7d, 24h, 1w, 30m)")
+ cmd.Flags().IntVar(&concurrency, "concurrency", 4, "Number of parallel sync workers")
+ cmd.Flags().StringVar(&dbPath, "db", "", "Database path (default: ~/.local/share/sync-walker-golden-pp-cli/data.db)")
+ cmd.Flags().IntVar(&maxPages, "max-pages", 100, "Maximum pages to fetch per resource (0 = unlimited; cap-hit emits a sync_warning event)")
+ cmd.Flags().BoolVar(&latestOnly, "latest-only", false, "Refresh head of each resource only; clears resume cursor and caps pages at 1. Mutually exclusive with --since (--since wins).")
+ cmd.Flags().BoolVar(&strict, "strict", false, "Exit non-zero on any per-resource failure (default: only critical failures or all-resource failure exit non-zero).")
+
+ return cmd
+}
+
+// syncResource handles the full paginated sync of a single resource.
+// It resumes from the last cursor unless sinceTS or full mode overrides it.
+func syncResource(c interface {
+ Get(string, map[string]string) (json.RawMessage, error)
+ RateLimit() float64
+}, db *store.Store, resource, sinceTS string, full bool, maxPages int) syncResult {
+ started := time.Now()
+
+ if !humanFriendly {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_start","resource":"%s"}`+"\n", resource)
+ }
+
+ path, err := syncResourcePath(resource)
+ if err != nil {
+ return syncResult{Resource: resource, Err: err, Duration: time.Since(started)}
+ }
+
+ // Skip resources whose path template still contains unresolved `{key}`
+ // placeholders after syncResourcePath() resolution. These paths require
+ // parent context (league_key, team_key, channel_id, etc.) that flat-list
+ // sync cannot fill. Emit a sync_warning describing the missing keys and
+ // continue — sync exits 0 if any resource succeeded, so this keeps
+ // hierarchical-API CLIs functional for the resources they CAN sync flat.
+ if missingKeys := unresolvedPathKeyRE.FindAllString(path, -1); len(missingKeys) > 0 {
+ if !humanFriendly {
+ payload := struct {
+ Event string `json:"event"`
+ Resource string `json:"resource"`
+ Reason string `json:"reason"`
+ Keys []string `json:"keys"`
+ Path string `json:"path"`
+ Message string `json:"message"`
+ }{
+ Event: "sync_warning",
+ Resource: resource,
+ Reason: "unfilled_path_key",
+ Keys: missingKeys,
+ Path: path,
+ Message: fmt.Sprintf("path %s requires parent context (%s); resource skipped", path, strings.Join(missingKeys, ", ")),
+ }
+ payloadJSON, _ := json.Marshal(payload)
+ fmt.Fprintf(os.Stdout, "%s\n", payloadJSON)
+ } else {
+ fmt.Fprintf(os.Stderr, " %s skipped (requires parent context: %s)\n",
+ resource, strings.Join(missingKeys, ", "))
+ }
+ return syncResult{
+ Resource: resource,
+ Warn: fmt.Errorf("skipped %s: unresolved path keys %v", resource, missingKeys),
+ Duration: time.Since(started),
+ }
+ }
+
+ var totalCount int
+
+ // Resume cursor from sync_state (unless --full cleared it)
+ existingCursor, lastSynced, _, _ := db.GetSyncState(resource)
+
+ // Determine the since param value:
+ // 1. Explicit --since flag takes priority
+ // 2. Otherwise use last_synced_at from sync_state for incremental sync
+ sinceParam := syncResourceSinceParam(resource)
+ effectiveSince := sinceTS
+ if effectiveSince == "" && !lastSynced.IsZero() && !full {
+ effectiveSince = lastSynced.Format(time.RFC3339)
+ }
+ // Resources whose list endpoint declares no temporal-filter parameter
+ // fall back to plain pagination — sending a synthetic since=... would
+ // reach the API as an unknown query param and (for strict APIs like
+ // Notion) fail the whole resource with a 400. Warn once per resource
+ // when the user expected incremental behavior.
+ if effectiveSince != "" && sinceParam == "" {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, " %s: incremental sync ignored (endpoint declares no temporal filter; falling back to full pagination)\n", resource)
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_warning","resource":"%s","reason":"resource_not_incremental","message":"endpoint does not declare a temporal filter parameter; incremental sync has no effect for this resource"}`+"\n", resource)
+ }
+ effectiveSince = ""
+ }
+
+ cursor := existingCursor
+ pageSize := determinePaginationDefaults()
+
+ var progressCount int64
+ pagesFetched := 0
+ lastNextCursor := ""
+ // extractFailureTotal accumulates per-item primary-key extraction
+ // misses across pages within this resource sync. Resource-level
+ // concurrency is 1 (one goroutine per resource via the work channel)
+ // so this counter cannot race. We emit one primary_key_unresolved
+ // sync_anomaly per resource per run when there's at least one miss
+ // (rate-limited via the anomalyEmitted flag) and a roll-up
+ // all_items_failed_id_extraction event when 100% of a single page
+ // failed extraction.
+ var extractFailureTotal int
+ var consumedTotal int
+ anomalyEmitted := false
+
+ for {
+ params := map[string]string{}
+
+ // Set page size
+ params[pageSize.limitParam] = strconv.Itoa(pageSize.limit)
+
+ // Set cursor for resume
+ if cursor != "" {
+ params[pageSize.cursorParam] = cursor
+ }
+
+ // Set since filter
+ if effectiveSince != "" {
+ params[sinceParam] = effectiveSince
+ }
+
+ data, err := c.Get(path, params)
+ if err != nil {
+ if w, ok := isSyncAccessWarning(err); ok {
+ if !humanFriendly {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_warning","resource":"%s","status":%d,"reason":"%s","message":"%s"}`+"\n",
+ resource, w.Status, w.Reason, strings.ReplaceAll(w.Message, `"`, `\"`))
+ }
+ return syncResult{Resource: resource, Count: totalCount, Warn: fmt.Errorf("skipped %s: %s", resource, w.Reason), Duration: time.Since(started)}
+ }
+ if !humanFriendly {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_error","resource":"%s","error":"%s"}`+"\n", resource, strings.ReplaceAll(err.Error(), `"`, `\"`))
+ }
+ return syncResult{Resource: resource, Count: totalCount, Err: fmt.Errorf("fetching %s: %w", resource, err), Duration: time.Since(started)}
+ }
+
+ // Try to extract items from the response.
+ // Strategy: try array first, then common wrapper keys.
+ items, nextCursor, hasMore := extractPageItems(data, pageSize.cursorParam)
+
+ if len(items) == 0 {
+ if isEmptyPageResponse(data) {
+ break
+ }
+ // Single object response - try to store as-is
+ if err := upsertSingleObject(db, resource, data); err != nil {
+ if !humanFriendly {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_error","resource":"%s","error":"%s"}`+"\n", resource, strings.ReplaceAll(err.Error(), `"`, `\"`))
+ }
+ return syncResult{Resource: resource, Err: err, Duration: time.Since(started)}
+ }
+ totalCount++
+ break
+ }
+
+ // Batch upsert all items from this page. UpsertBatch returns
+ // (stored, extractFailures, err): stored counts rows actually
+ // landed; extractFailures counts items that survived JSON
+ // unmarshal but had no extractable primary key (templated
+ // IDField AND generic fallback both missed). Tracking these
+ // separately lets us emit precise sync_anomaly events: a
+ // roll-up "all_items_failed_id_extraction" when an entire
+ // page yields zero stored, a per-resource
+ // "primary_key_unresolved" the first time any single item
+ // fails, and the F4b "stored_count_zero_after_extraction"
+ // probe when extraction succeeded but rows still didn't land.
+ stored, extractFailures, err := upsertResourceBatch(db, resource, items)
+ if err != nil {
+ if !humanFriendly {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_error","resource":"%s","error":"%s"}`+"\n", resource, strings.ReplaceAll(err.Error(), `"`, `\"`))
+ }
+ return syncResult{Resource: resource, Count: totalCount, Err: fmt.Errorf("upserting batch for %s: %w", resource, err), Duration: time.Since(started)}
+ }
+
+ consumedTotal += len(items)
+ extractFailureTotal += extractFailures
+
+ // When a non-empty page yielded zero stored rows, the API
+ // returned items in a shape we couldn't extract IDs from —
+ // likely scalar IDs (Firebase /topstories.json, GitHub user-
+ // repo lists) where the spec author should declare a hydration
+ // pattern, or an unrecognized primary-key field name.
+ if len(items) > 0 && stored == 0 {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "warning: %s returned %d items but stored 0 — the local store will be empty for this resource. Likely cause: scalar item shape rather than objects with extractable IDs.\n", resource, len(items))
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_anomaly","resource":"%s","consumed":%d,"stored":0,"reason":"all_items_failed_id_extraction"}`+"\n", resource, len(items))
+ }
+ anomalyEmitted = true
+ } else if extractFailures > 0 && !anomalyEmitted {
+ // Per-item primary-key resolution failure but at least one
+ // item landed — emit one structured warning per resource per
+ // sync run so users see the first occurrence of silent drops
+ // instead of waiting for total failure.
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\nwarning: %s had %d item(s) on this page with no extractable primary key — those rows were dropped silently. Annotate the spec with x-resource-id to fix.\n", resource, extractFailures)
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_anomaly","resource":"%s","consumed":%d,"stored":%d,"count":%d,"reason":"primary_key_unresolved"}`+"\n", resource, len(items), stored, extractFailures)
+ }
+ anomalyEmitted = true
+ }
+
+ totalCount += stored
+ atomic.AddInt64(&progressCount, int64(stored))
+
+ // Progress reporting (include rate limit info when active)
+ currentRate := c.RateLimit()
+ if humanFriendly {
+ if currentRate > 0 {
+ fmt.Fprintf(os.Stderr, "\r %s: %d synced [%.1f req/s]", resource, atomic.LoadInt64(&progressCount), currentRate)
+ } else {
+ fmt.Fprintf(os.Stderr, "\r %s: %d synced", resource, atomic.LoadInt64(&progressCount))
+ }
+ } else {
+ if currentRate > 0 {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_progress","resource":"%s","fetched":%d,"rate_rps":%.1f}`+"\n", resource, atomic.LoadInt64(&progressCount), currentRate)
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_progress","resource":"%s","fetched":%d}`+"\n", resource, atomic.LoadInt64(&progressCount))
+ }
+ }
+
+ // Save cursor after each page for resumability
+ if err := db.SaveSyncState(resource, nextCursor, totalCount); err != nil {
+ // Non-fatal: log and continue
+ fmt.Fprintf(os.Stderr, "\nwarning: failed to save sync state for %s: %v\n", resource, err)
+ }
+
+ pagesFetched++
+
+ // Enforce page ceiling to prevent runaway syncs on large-catalog APIs
+ if maxPages > 0 && pagesFetched >= maxPages {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\n %s: reached --max-pages limit (%d pages, %d items)\n", resource, maxPages, totalCount)
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_warning","resource":"%s","reason":"max_pages_cap_hit","message":"reached --max-pages cap of %d; data may be truncated. Re-run with --max-pages 0 (unlimited) or higher to verify."}`+"\n", resource, maxPages)
+ }
+ break
+ }
+
+ // Sticky-cursor detector: if the API echoes the same next cursor across
+ // consecutive pages without advancing, abort to prevent burning the
+ // --max-pages budget on a non-terminating loop. Checked AFTER the cap
+ // guard so cap-hit takes precedence; checked BEFORE the natural-end
+ // check below because the natural-end check would not catch a sticky
+ // non-empty cursor on its own.
+ if nextCursor != "" && nextCursor == lastNextCursor {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\n %s: API returned the same next cursor across two pages; aborting to prevent budget waste.\n", resource)
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_warning","resource":"%s","reason":"stuck_pagination","message":"API returned the same next cursor across two pages for resource %s; aborting to prevent budget waste."}`+"\n", resource, resource)
+ }
+ break
+ }
+ lastNextCursor = nextCursor
+
+ // Determine if there are more pages
+ if !hasMore || len(items) < pageSize.limit || nextCursor == "" {
+ break
+ }
+
+ cursor = nextCursor
+ }
+
+ // Final sync state: clear cursor (sync is complete), update count
+ _ = db.SaveSyncState(resource, "", totalCount)
+
+ // F4b symptom probe: if items were consumed and successfully
+ // extracted (extractFailures < consumed) but nothing landed in
+ // the store, something downstream of extraction silently dropped
+ // rows — FTS5 trigger error, transaction rollback, character
+ // encoding. Emit a sync_anomaly so the symptom is visible the
+ // next time it recurs; the underlying root cause is held out for
+ // controlled repro.
+ if consumedTotal > 0 && totalCount == 0 && extractFailureTotal < consumedTotal {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\nwarning: %s consumed %d items, extracted %d primary keys, but stored 0 rows — extraction succeeded yet nothing landed. Investigate FTS triggers / transaction rollback / encoding.\n", resource, consumedTotal, consumedTotal-extractFailureTotal)
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_anomaly","resource":"%s","consumed":%d,"stored":0,"extract_failures":%d,"reason":"stored_count_zero_after_extraction"}`+"\n", resource, consumedTotal, extractFailureTotal)
+ }
+ }
+
+ if !humanFriendly {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_complete","resource":"%s","total":%d,"duration_ms":%d}`+"\n", resource, totalCount, time.Since(started).Milliseconds())
+ }
+
+ return syncResult{Resource: resource, Count: totalCount, Duration: time.Since(started)}
+}
+
+// paginationDefaults holds the resolved pagination parameter names and page size.
+type paginationDefaults struct {
+ cursorParam string
+ limitParam string
+ limit int
+}
+
+// determinePaginationDefaults returns the pagination parameter names to use.
+// Values are detected from the API spec by the profiler at generation time.
+func determinePaginationDefaults() paginationDefaults {
+ return paginationDefaults{
+ cursorParam: "after",
+ limitParam: "limit",
+ limit: 100,
+ }
+}
+
+// syncResourceSinceParam returns the query parameter name this resource's
+// list endpoint declares for incremental temporal filtering, or "" when the
+// endpoint declares none. Skipping the param for "" resources avoids
+// validation-error 400s on APIs that reject unknown query keys.
+func syncResourceSinceParam(resource string) string {
+ switch resource {
+ }
+ return ""
+}
+
+// extractPageItems attempts to extract an array of items and pagination cursor from a response.
+// It tries multiple strategies:
+// 1. Direct JSON array
+// 2. Common wrapper keys: "data", "results", "items", "records", "nodes", "entries"
+// It also extracts the next cursor from common response fields.
+func extractPageItems(data json.RawMessage, cursorParam string) ([]json.RawMessage, string, bool) {
+ // Strategy 1: direct array
+ var items []json.RawMessage
+ if err := json.Unmarshal(data, &items); err == nil {
+ return items, "", false
+ }
+
+ // Strategy 2: object with known wrapper keys
+ var envelope map[string]json.RawMessage
+ if err := json.Unmarshal(data, &envelope); err != nil {
+ return nil, "", false
+ }
+
+ // Try common item keys first (fast path)
+ for _, key := range pageItemKeys {
+ if raw, ok := envelope[key]; ok {
+ if err := json.Unmarshal(raw, &items); err == nil && len(items) > 0 {
+ nextCursor, hasMore := extractPaginationFromEnvelope(envelope, cursorParam)
+ return items, nextCursor, hasMore
+ }
+ }
+ }
+
+ // Fallback: try every key in the envelope. If exactly one maps to a JSON
+ // array with items, use it. This handles APIs that wrap responses with the
+ // resource name (e.g., {"markets": [...], "cursor": "..."}).
+ var arrayKey string
+ var arrayItems []json.RawMessage
+ arrayCount := 0
+ for key, raw := range envelope {
+ var candidate []json.RawMessage
+ if err := json.Unmarshal(raw, &candidate); err == nil && len(candidate) > 0 {
+ arrayKey = key
+ arrayItems = candidate
+ arrayCount++
+ }
+ }
+ if arrayCount == 1 {
+ nextCursor, hasMore := extractPaginationFromEnvelope(envelope, cursorParam)
+ _ = arrayKey // used for detection, items extracted above
+ return arrayItems, nextCursor, hasMore
+ }
+
+ return nil, "", false
+}
+
+func isEmptyPageResponse(data json.RawMessage) bool {
+ var direct []json.RawMessage
+ if err := json.Unmarshal(data, &direct); err == nil && !isJSONNull(data) {
+ return len(direct) == 0
+ }
+
+ var envelope map[string]json.RawMessage
+ if err := json.Unmarshal(data, &envelope); err != nil {
+ return false
+ }
+
+ for _, key := range pageItemKeys {
+ if raw, ok := envelope[key]; ok {
+ var items []json.RawMessage
+ if err := json.Unmarshal(raw, &items); err == nil && !isJSONNull(raw) {
+ return len(items) == 0
+ }
+ }
+ }
+
+ arrayCount := 0
+ for _, raw := range envelope {
+ var candidate []json.RawMessage
+ if err := json.Unmarshal(raw, &candidate); err == nil && len(candidate) == 0 && !isJSONNull(raw) {
+ arrayCount++
+ }
+ }
+ return arrayCount == 1
+}
+
+func isJSONNull(raw json.RawMessage) bool {
+ return strings.TrimSpace(string(raw)) == "null"
+}
+
+// extractPaginationFromEnvelope extracts cursor and has_more from a response envelope.
+func extractPaginationFromEnvelope(envelope map[string]json.RawMessage, cursorParam string) (string, bool) {
+ var hasMore bool
+
+ nextCursor := nextCursorFromLinks(envelope, cursorParam)
+
+ // Try common cursor field names
+ cursorKeys := []string{
+ "next_cursor", "nextCursor", "cursor", "next_page_token",
+ "nextPageToken", "page_token", "after", "end_cursor", "endCursor",
+ }
+ if nextCursor == "" {
+ nextCursor = findCursorInMap(envelope, cursorKeys)
+ }
+
+ // If no top-level cursor was found, look one level deeper into well-known
+ // pagination wrapper objects. Slack returns {"messages":[...],
+ // "response_metadata":{"next_cursor":"..."}}; MongoDB Atlas uses
+ // "pagination"; many APIs use "meta" or "paging". Purely additive — only
+ // runs when the top-level scan returned empty — and uses the same
+ // cursorKeys set so wrapper contents go through the same name match.
+ if nextCursor == "" {
+ paginationWrapperKeys := []string{"response_metadata", "pagination", "meta", "paging"}
+ for _, wrapperKey := range paginationWrapperKeys {
+ rawWrapper, ok := envelope[wrapperKey]
+ if !ok {
+ continue
+ }
+ var inner map[string]json.RawMessage
+ if json.Unmarshal(rawWrapper, &inner) != nil {
+ continue
+ }
+ if c := findCursorInMap(inner, cursorKeys); c != "" {
+ nextCursor = c
+ break
+ }
+ }
+ }
+
+ // Try common has_more field names
+ hasMoreKeys := []string{"has_more", "hasMore", "has_next", "hasNext", "next_page"}
+ for _, key := range hasMoreKeys {
+ if raw, ok := envelope[key]; ok {
+ if err := json.Unmarshal(raw, &hasMore); err == nil {
+ break
+ }
+ }
+ }
+
+ // If we found a cursor, assume there are more pages even without explicit has_more
+ if nextCursor != "" && !hasMore {
+ hasMore = true
+ }
+
+ return nextCursor, hasMore
+}
+
+// nextCursorFromLinks extracts JSON:API-style pagination cursors from
+// {"links":{"next":"https://example.com/items?page[cursor]=..."}}.
+func nextCursorFromLinks(envelope map[string]json.RawMessage, cursorParam string) string {
+ rawLinks, ok := envelope["links"]
+ if !ok {
+ return ""
+ }
+ var links map[string]json.RawMessage
+ if json.Unmarshal(rawLinks, &links) != nil {
+ return ""
+ }
+ rawNext, ok := links["next"]
+ if !ok {
+ return ""
+ }
+ var nextURL string
+ if json.Unmarshal(rawNext, &nextURL) != nil || nextURL == "" {
+ return ""
+ }
+
+ cursorKeys := []string{cursorParam}
+ if cursorParam != "page[cursor]" {
+ cursorKeys = append(cursorKeys, "page[cursor]")
+ }
+ if cursorParam != "cursor" {
+ cursorKeys = append(cursorKeys, "cursor")
+ }
+ if cursorParam != "after" {
+ cursorKeys = append(cursorKeys, "after")
+ }
+
+ parsed, err := url.Parse(nextURL)
+ if err != nil {
+ return ""
+ }
+ values := parsed.Query()
+ for _, key := range cursorKeys {
+ if key == "" {
+ continue
+ }
+ if cursor := values.Get(key); cursor != "" {
+ return cursor
+ }
+ }
+ return ""
+}
+
+// findCursorInMap returns the first non-empty string-typed value in m
+// whose key matches one of cursorKeys. Used by extractPaginationFromEnvelope
+// to scan both the top-level envelope and well-known wrapper objects with
+// the same name-match rules — extracted so the two scans can't drift.
+func findCursorInMap(m map[string]json.RawMessage, cursorKeys []string) string {
+ for _, key := range cursorKeys {
+ raw, ok := m[key]
+ if !ok {
+ continue
+ }
+ var s string
+ if err := json.Unmarshal(raw, &s); err == nil && s != "" {
+ return s
+ }
+ }
+ return ""
+}
+
+type discriminatorDispatch struct {
+ Field string
+ Values map[string]string
+}
+
+var discriminatorDispatchers = map[string]discriminatorDispatch{
+}
+
+func upsertResourceBatch(db *store.Store, resource string, items []json.RawMessage) (int, int, error) {
+ if _, ok := discriminatorDispatchers[resource]; !ok {
+ return db.UpsertBatch(resource, items)
+ }
+
+ grouped := map[string][]json.RawMessage{}
+ order := []string{}
+ for _, item := range items {
+ target := resource
+ var obj map[string]any
+ if err := json.Unmarshal(item, &obj); err == nil {
+ target = resolveDiscriminatedResource(resource, obj)
+ }
+ if _, ok := grouped[target]; !ok {
+ order = append(order, target)
+ }
+ grouped[target] = append(grouped[target], item)
+ }
+
+ var stored, extractFailures int
+ for _, target := range order {
+ targetStored, targetExtractFailures, err := db.UpsertBatch(target, grouped[target])
+ if err != nil {
+ return stored, extractFailures + targetExtractFailures, err
+ }
+ stored += targetStored
+ extractFailures += targetExtractFailures
+ }
+ return stored, extractFailures, nil
+}
+
+func resolveDiscriminatedResource(resource string, obj map[string]any) string {
+ dispatcher, ok := discriminatorDispatchers[resource]
+ if !ok || dispatcher.Field == "" {
+ return resource
+ }
+ value := store.LookupFieldValue(obj, dispatcher.Field)
+ if value == nil {
+ return resource
+ }
+ if target, ok := dispatcher.Values[fmt.Sprintf("%v", value)]; ok && target != "" {
+ return target
+ }
+ return resource
+}
+
+// upsertSingleObject stores a non-array API response as a single record.
+func upsertSingleObject(db *store.Store, resource string, data json.RawMessage) error {
+ var obj map[string]any
+ if err := json.Unmarshal(data, &obj); err != nil {
+ // Not a JSON object either - store raw under resource name
+ return db.Upsert(resource, resource, data)
+ }
+
+ resource = resolveDiscriminatedResource(resource, obj)
+
+ id := extractID(resource, obj)
+ if id == "" {
+ id = resource
+ }
+
+ switch resource {
+ case "leagues":
+ return db.UpsertLeagues(data)
+ default:
+ return db.Upsert(resource, id, data)
+ }
+}
+
+// parseSinceDuration converts human-friendly duration strings into a time.Time.
+// Supported formats: "7d" (days), "24h" (hours), "30m" (minutes), "1w" (weeks).
+func parseSinceDuration(s string) (time.Time, error) {
+ re := regexp.MustCompile(`^(\d+)([dhwm])$`)
+ matches := re.FindStringSubmatch(strings.TrimSpace(s))
+ if matches == nil {
+ return time.Time{}, fmt.Errorf("expected format like 7d, 24h, 1w, or 30m")
+ }
+
+ n, err := strconv.Atoi(matches[1])
+ if err != nil {
+ return time.Time{}, err
+ }
+
+ now := time.Now()
+ switch matches[2] {
+ case "d":
+ return now.Add(-time.Duration(n) * 24 * time.Hour), nil
+ case "h":
+ return now.Add(-time.Duration(n) * time.Hour), nil
+ case "w":
+ return now.Add(-time.Duration(n) * 7 * 24 * time.Hour), nil
+ case "m":
+ return now.Add(-time.Duration(n) * time.Minute), nil
+ default:
+ return time.Time{}, fmt.Errorf("unknown unit %q", matches[2])
+ }
+}
+
+func defaultSyncResources() []string {
+ return []string{
+ "games",
+ }
+}
+
+// syncResourcePath maps resource names to their actual API endpoint paths.
+// For REST APIs this is typically "/<resource>". For non-REST APIs (e.g., Steam)
+// this preserves the actual endpoint path like "/ISteamApps/GetAppList/v2".
+func syncResourcePath(resource string) (string, error) {
+ paths := map[string]string{
+ "games": "/games",
+ }
+ if p, ok := paths[resource]; ok {
+ return p, nil
+ }
+ return "", fmt.Errorf("unknown sync resource %q", resource)
+}
+
+// dependentResourceDef describes a child resource that requires iterating parent IDs to sync.
+// When KeyField is non-empty, the dependent's parent IDs are extracted from the parent
+// records' KeyField via json_extract rather than from the parent table's primary key.
+// Populated from a spec-declared walker (Endpoint.Walker.KeyField in internal YAML,
+// or `key_field` under `x-pp-sync-walker` in OpenAPI). Empty KeyField preserves the
+// existing parent-primary-key flow byte-for-byte.
+type dependentResourceDef struct {
+ Name string
+ ParentTable string
+ ParentIDParam string
+ PathTemplate string
+ KeyField string
+}
+
+func dependentResourceDefs() []dependentResourceDef {
+ return []dependentResourceDef{
+ {Name: "leagues", ParentTable: "games", ParentIDParam: "game_key", PathTemplate: "/games/{game_key}/leagues", KeyField: "game_key"},
+ }
+}
+
+// syncDependentResources iterates parent tables and syncs child resources per parent ID.
+// parentFilter mirrors the user's --resources flag (empty = sync everything). A dependent
+// runs when its parent table or its own name appears in the filter.
+func syncDependentResources(c interface {
+ Get(string, map[string]string) (json.RawMessage, error)
+ RateLimit() float64
+}, db *store.Store, sinceTS string, full bool, maxPages int, parentFilter []string) []syncResult {
+ allow := make(map[string]bool, len(parentFilter))
+ for _, r := range parentFilter {
+ allow[r] = true
+ }
+ var results []syncResult
+ for _, dep := range dependentResourceDefs() {
+ if len(allow) > 0 && !allow[dep.ParentTable] && !allow[dep.Name] {
+ continue
+ }
+ res := syncDependentResource(c, db, dep, sinceTS, full, maxPages)
+ results = append(results, res)
+ }
+ return results
+}
+
+// syncDependentResource syncs a single child resource by iterating all parent IDs.
+func syncDependentResource(c interface {
+ Get(string, map[string]string) (json.RawMessage, error)
+ RateLimit() float64
+}, db *store.Store, dep dependentResourceDef, sinceTS string, full bool, maxPages int) syncResult {
+ started := time.Now()
+
+ // Query parent table for the keys to substitute into the child path.
+ // When KeyField is empty, use the parent's primary key via ListIDs
+ // (the original flat parent-child flow). When KeyField is set, the
+ // spec declared a walker that extracts a non-PK field from each parent
+ // record — ListField looks up the field in the parent's typed column
+ // if present, otherwise json_extract from the generic resources table.
+ var parentIDs []string
+ var err error
+ if dep.KeyField != "" {
+ parentIDs, err = db.ListField(dep.ParentTable, dep.KeyField)
+ } else {
+ parentIDs, err = db.ListIDs(dep.ParentTable)
+ }
+ if err != nil || len(parentIDs) == 0 {
+ if len(parentIDs) == 0 {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, " %s: skipping (parent table %s is empty, sync it first)\n", dep.Name, dep.ParentTable)
+ }
+ return syncResult{Resource: dep.Name, Duration: time.Since(started)}
+ }
+ return syncResult{Resource: dep.Name, Err: fmt.Errorf("querying parent table %s: %w", dep.ParentTable, err), Duration: time.Since(started)}
+ }
+
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, " %s: syncing for %d %s parents\n", dep.Name, len(parentIDs), dep.ParentTable)
+ }
+
+ var totalCount int
+ var deniedParents int
+ var firstDenial *accessWarning
+ pageSize := determinePaginationDefaults()
+ depSinceParam := syncResourceSinceParam(dep.Name)
+ depSinceTS := sinceTS
+ if depSinceTS != "" && depSinceParam == "" {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, " %s: incremental sync ignored (endpoint declares no temporal filter; falling back to full pagination)\n", dep.Name)
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_warning","resource":"%s","reason":"resource_not_incremental","message":"endpoint does not declare a temporal filter parameter; incremental sync has no effect for this resource"}`+"\n", dep.Name)
+ }
+ depSinceTS = ""
+ }
+ // Per-resource extract-failure tracking for the F4b symptom probe and
+ // per-item primary_key_unresolved warning. See syncResource for the
+ // concurrency rationale (one goroutine per resource → no race).
+ var depExtractFailureTotal int
+ var depConsumedTotal int
+ depAnomalyEmitted := false
+
+ for idx, parentID := range parentIDs {
+ // Build child endpoint path by replacing the param placeholder
+ path := strings.Replace(dep.PathTemplate, "{"+dep.ParentIDParam+"}", parentID, 1)
+
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\r %s: syncing for %s (%d/%d parents)", dep.Name, dep.ParentTable, idx+1, len(parentIDs))
+ }
+
+ cursor := ""
+ pagesFetched := 0
+ lastNextCursor := ""
+
+ for {
+ params := map[string]string{}
+ params[pageSize.limitParam] = strconv.Itoa(pageSize.limit)
+ if cursor != "" {
+ params[pageSize.cursorParam] = cursor
+ }
+ if depSinceTS != "" {
+ params[depSinceParam] = depSinceTS
+ }
+
+ data, err := c.Get(path, params)
+ if err != nil {
+ // Non-fatal per parent: log and continue to next parent.
+ // Track access-denial separately so an all-denied dependent
+ // resource can surface as a Warn rather than silent success.
+ if w, ok := isSyncAccessWarning(err); ok {
+ deniedParents++
+ if firstDenial == nil {
+ firstDenial = w
+ }
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\n %s: access denied for parent %s: %s\n", dep.Name, parentID, w.Reason)
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_warning","resource":"%s","parent":"%s","status":%d,"reason":"%s","message":"%s"}`+"\n",
+ dep.Name, parentID, w.Status, w.Reason, strings.ReplaceAll(w.Message, `"`, `\"`))
+ }
+ } else if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\n %s: error for parent %s: %v\n", dep.Name, parentID, err)
+ }
+ break
+ }
+
+ items, nextCursor, hasMore := extractPageItems(data, pageSize.cursorParam)
+ if len(items) == 0 {
+ break
+ }
+
+ // Inject parent_id into each item before upserting
+ for i, item := range items {
+ var obj map[string]json.RawMessage
+ if err := json.Unmarshal(item, &obj); err == nil {
+ parentIDJSON, _ := json.Marshal(parentID)
+ obj["parent_id"] = parentIDJSON
+ if modified, err := json.Marshal(obj); err == nil {
+ items[i] = modified
+ }
+ }
+ }
+
+ stored, extractFailures, err := upsertResourceBatch(db, dep.Name, items)
+ if err != nil {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\n %s: upsert error for parent %s: %v\n", dep.Name, parentID, err)
+ }
+ break
+ }
+
+ depConsumedTotal += len(items)
+ depExtractFailureTotal += extractFailures
+ // Order matches the flat path (syncResource): all-fail first,
+ // then partial-fail. The all-fail case dominates — if stored==0
+ // with at least one item, the resource is broken regardless of
+ // how many extractions failed individually.
+ if len(items) > 0 && stored == 0 && !depAnomalyEmitted {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\nwarning: %s returned %d items for parent %s but stored 0 — likely scalar item shape or unrecognized primary-key field name.\n", dep.Name, len(items), parentID)
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_anomaly","resource":"%s","parent":"%s","consumed":%d,"stored":0,"reason":"all_items_failed_id_extraction"}`+"\n", dep.Name, parentID, len(items))
+ }
+ depAnomalyEmitted = true
+ } else if extractFailures > 0 && stored > 0 && !depAnomalyEmitted {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\nwarning: %s had %d item(s) with no extractable primary key — those rows were dropped silently. Annotate the spec with x-resource-id to fix.\n", dep.Name, extractFailures)
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_anomaly","resource":"%s","parent":"%s","consumed":%d,"stored":%d,"count":%d,"reason":"primary_key_unresolved"}`+"\n", dep.Name, parentID, len(items), stored, extractFailures)
+ }
+ depAnomalyEmitted = true
+ }
+
+ totalCount += stored
+ pagesFetched++
+
+ if maxPages > 0 && pagesFetched >= maxPages {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\n %s: reached --max-pages limit (%d pages, %d items) for parent %s\n", dep.Name, maxPages, totalCount, parentID)
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_warning","resource":"%s","parent":"%s","reason":"max_pages_cap_hit","message":"reached --max-pages cap of %d; data may be truncated. Re-run with --max-pages 0 (unlimited) or higher to verify."}`+"\n", dep.Name, parentID, maxPages)
+ }
+ break
+ }
+ // Sticky-cursor detector: see syncResource for rationale. Same shape
+ // here so dependent-resource page loops cannot burn the budget on a
+ // non-advancing next cursor.
+ if nextCursor != "" && nextCursor == lastNextCursor {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\n %s: API returned the same next cursor across two pages for parent %s; aborting to prevent budget waste.\n", dep.Name, parentID)
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_warning","resource":"%s","parent":"%s","reason":"stuck_pagination","message":"API returned the same next cursor across two pages for resource %s; aborting to prevent budget waste."}`+"\n", dep.Name, parentID, dep.Name)
+ }
+ break
+ }
+ lastNextCursor = nextCursor
+ if !hasMore || len(items) < pageSize.limit || nextCursor == "" {
+ break
+ }
+ cursor = nextCursor
+ }
+
+ // Brief rate-limit pause between parents to avoid hammering the API
+ time.Sleep(100 * time.Millisecond)
+ }
+
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\n")
+ }
+
+ _ = db.SaveSyncState(dep.Name, "", totalCount)
+
+ // F4b symptom probe: items consumed and extracted but nothing landed.
+ // See syncResource for rationale.
+ if depConsumedTotal > 0 && totalCount == 0 && depExtractFailureTotal < depConsumedTotal {
+ if humanFriendly {
+ fmt.Fprintf(os.Stderr, "\nwarning: %s consumed %d items, extracted %d primary keys, but stored 0 rows — extraction succeeded yet nothing landed. Investigate FTS triggers / transaction rollback / encoding.\n", dep.Name, depConsumedTotal, depConsumedTotal-depExtractFailureTotal)
+ } else {
+ fmt.Fprintf(os.Stdout, `{"event":"sync_anomaly","resource":"%s","consumed":%d,"stored":0,"extract_failures":%d,"reason":"stored_count_zero_after_extraction"}`+"\n", dep.Name, depConsumedTotal, depExtractFailureTotal)
+ }
+ }
+
+ // If every parent was access-denied and nothing was synced, surface as a
+ // warning so the run-level summary and exit code reflect insufficient access.
+ if deniedParents == len(parentIDs) && totalCount == 0 && firstDenial != nil {
+ return syncResult{
+ Resource: dep.Name,
+ Count: 0,
+ Warn: fmt.Errorf("skipped %s: %s on all %d parents", dep.Name, firstDenial.Reason, len(parentIDs)),
+ Duration: time.Since(started),
+ }
+ }
+ return syncResult{Resource: dep.Name, Count: totalCount, Duration: time.Since(started)}
+}
+
+// resourceIDFieldOverrides projects per-resource IDField (set by the profiler
+// from x-resource-id or the response-schema fallback chain) into a runtime
+// lookup map. extractID consults this first so the templated path wins over
+// the generic fallback list; the generic list applies only when the override
+// is empty or the override field is absent on a given item.
+//
+// Includes both flat resources and dependent (parent-child) resources so
+// annotations on a child path-item are honored at runtime, not just on
+// flat paths.
+var resourceIDFieldOverrides = map[string]string{
+ "games": "game_key",
+}
+
+// genericIDFieldFallbacks is the runtime safety net for resources that did
+// NOT receive a templated IDField. API-specific names belong in spec
+// annotations (x-resource-id), not this list.
+var genericIDFieldFallbacks = []string{"id", "ID", "name", "uuid", "slug", "key", "code", "uid"}
+
+var pageItemKeys = []string{"data", "results", "items", "records", "nodes", "entries"}
+
+// criticalResources is the template-time projection of per-resource Critical
+// (set by the profiler from the spec's path-item x-critical extension). It
+// is consulted at error-aggregation time so a non-critical failure can be
+// downgraded to a sync_warning + exit 0 unless --strict was passed.
+//
+// Includes both flat resources and dependent (parent-child) resources so a
+// failed child sync flagged x-critical: true exits non-zero just like a
+// flat-resource critical failure.
+var criticalResources = map[string]bool{
+}
+
+// extractID resolves an item's primary-key field. It consults the
+// per-resource templated override first; on miss, it falls through to the
+// generic fallback list. resource may be empty for callers that don't have
+// a resource context (only the generic list applies in that case).
+//
+// Field lookups go through store.LookupFieldValue so snake_case overrides
+// match camelCase JSON renderings. UpsertBatch resolves fields the same
+// way — divergence between the two paths produces silent drops on
+// heterogeneous payloads.
+func extractID(resource string, obj map[string]any) string {
+ if override, ok := resourceIDFieldOverrides[resource]; ok && override != "" {
+ if v := store.LookupFieldValue(obj, override); v != nil {
+ s := fmt.Sprintf("%v", v)
+ if s != "" && s != "<nil>" {
+ return s
+ }
+ }
+ }
+ for _, key := range genericIDFieldFallbacks {
+ if v := store.LookupFieldValue(obj, key); v != nil {
+ s := fmt.Sprintf("%v", v)
+ if s != "" && s != "<nil>" {
+ return s
+ }
+ }
+ }
+ return ""
+}
diff --git a/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/store/store.go b/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/store/store.go
new file mode 100644
index 00000000..21316528
--- /dev/null
+++ b/testdata/golden/expected/generate-sync-walker-api/sync-walker-golden/internal/store/store.go
@@ -0,0 +1,1150 @@
+// Copyright 2026 printing-press-golden. Licensed under Apache-2.0. See LICENSE.
+// Generated by CLI Printing Press (https://github.com/mvanhorn/cli-printing-press). DO NOT EDIT.
+
+// Package store provides local SQLite persistence for sync-walker-golden-pp-cli.
+// Uses modernc.org/sqlite (pure Go, no CGO) for zero-dependency cross-compilation.
+// FTS5 full-text search indexes are created for searchable content.
+package store
+
+import (
+ "context"
+ "database/sql"
+ "encoding/json"
+ "fmt"
+ "os"
+ "path/filepath"
+ "regexp"
+ "strings"
+ "sync"
+ "time"
+
+ _ "modernc.org/sqlite"
+)
+
+var uuidPattern = regexp.MustCompile(`^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$`)
+
+// validIdentifierRE pins ListField's `field` argument to a safe SQL
+// identifier shape before any Sprintf interpolation. Matches what
+// pragma_table_info implicitly enforces on the primary path, so the
+// fallback path inherits the same defense without depending on whether
+// the parent's typed domain table exists at the moment of the lookup.
+var validIdentifierRE = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`)
+
+// IsUUID returns true if the input looks like a UUID.
+func IsUUID(s string) bool {
+ return uuidPattern.MatchString(s)
+}
+
+// StoreSchemaVersion is the on-disk schema version this binary understands.
+// It is stamped into SQLite's PRAGMA user_version on fresh databases and
+// checked on every open. Bump this whenever a migration changes table
+// shape — adding columns, dropping indexes, changing FTS5 tokenizers —
+// so an older binary refuses to open a newer database rather than silently
+// producing wrong results against a schema it cannot read.
+const StoreSchemaVersion = 2
+
+const resourcesFTSCreateSQL = `CREATE VIRTUAL TABLE IF NOT EXISTS resources_fts USING fts5(
+ id, resource_type, content, tokenize='porter unicode61'
+)`
+
+type Store struct {
+ db *sql.DB
+ // writeMu serializes all DB writes. Read paths bypass the lock and run
+ // concurrently against WAL. Resource-level concurrency in sync.go.tmpl
+ // is 1 (one goroutine per resource via len(resources)-sized work channel)
+ // — read-then-write sequences (e.g., GetSyncCursor → SaveSyncState) are
+ // race-free by construction within a resource.
+ writeMu sync.Mutex
+ path string
+}
+
+// Open opens or creates the SQLite store at dbPath using the background
+// context. Prefer OpenWithContext from a Cobra command so SIGINT during
+// a slow migration interrupts the open instead of stranding the caller.
+func Open(dbPath string) (*Store, error) {
+ return OpenWithContext(context.Background(), dbPath)
+}
+
+// OpenReadOnly opens an existing SQLite store at dbPath in read-only mode.
+// mode=ro rejects direct and CTE-wrapped writes (INSERT, UPDATE, DELETE,
+// REPLACE, "WITH x AS (...) INSERT ...") at the driver level. Skips
+// MkdirAll and migrate; the file is expected to exist.
+//
+// The file: URI prefix is load-bearing: modernc.org/sqlite only honors
+// SQLite's URI query parameters (mode, cache, etc.) when the DSN starts
+// with "file:". Without the prefix, "?mode=ro" is silently dropped and
+// the connection opens read-write. Underscore-prefixed driver pragmas
+// (_journal_mode, _busy_timeout, etc.) work either way; they're parsed
+// out of the DSN by the driver before sqlite3_open_v2.
+func OpenReadOnly(dbPath string) (*Store, error) {
+ db, err := sql.Open("sqlite", "file:"+dbPath+"?mode=ro&_journal_mode=WAL&_busy_timeout=5000&_foreign_keys=ON&_temp_store=MEMORY&_mmap_size=268435456")
+ if err != nil {
+ return nil, fmt.Errorf("opening database (read-only): %w", err)
+ }
+ db.SetMaxOpenConns(2)
+ return &Store{db: db, path: dbPath}, nil
+}
+
+// OpenWithContext opens or creates the SQLite store at dbPath. The
+// context is honored by the migration path: cancellation interrupts the
+// retry-on-SQLITE_BUSY loop and propagates ctx.Err() back to the caller
+// instead of waiting out the full migrationLockTimeout.
+func OpenWithContext(ctx context.Context, dbPath string) (*Store, error) {
+ if err := os.MkdirAll(filepath.Dir(dbPath), 0o755); err != nil {
+ return nil, fmt.Errorf("creating db directory: %w", err)
+ }
+
+ db, err := sql.Open("sqlite", dbPath+"?_journal_mode=WAL&_synchronous=NORMAL&_busy_timeout=5000&_foreign_keys=ON&_temp_store=MEMORY&_mmap_size=268435456")
+ if err != nil {
+ return nil, fmt.Errorf("opening database: %w", err)
+ }
+
+ // WAL mode + 2 connections allows one read cursor open while a second
+ // query executes (e.g., analytics commands calling helpers during row
+ // iteration). Writes are still serialized by SQLite's WAL lock.
+ db.SetMaxOpenConns(2)
+
+ s := &Store{db: db, path: dbPath}
+ if err := s.migrate(ctx); err != nil {
+ db.Close()
+ return nil, fmt.Errorf("running migrations: %w", err)
+ }
+
+ return s, nil
+}
+
+func (s *Store) Close() error {
+ return s.db.Close()
+}
+
+// Path returns the on-disk path of the backing SQLite file.
+func (s *Store) Path() string {
+ return s.path
+}
+
+// DB exposes the underlying *sql.DB for callers that need to run ad-hoc
+// queries (e.g., doctor's cache inspection, share snapshot import).
+// Callers must not call Close on the returned handle.
+func (s *Store) DB() *sql.DB {
+ return s.db
+}
+
+// SchemaVersion reads PRAGMA user_version, which is stamped by migrate().
+// A zero value means the database predates the schema-version gate — not
+// a bug, but the caller may want to warn.
+func (s *Store) SchemaVersion() (int, error) {
+ var v int
+ if err := s.db.QueryRow(`PRAGMA user_version`).Scan(&v); err != nil {
+ return 0, fmt.Errorf("read user_version: %w", err)
+ }
+ return v, nil
+}
+
+// ensureColumn adds a column to an existing table if it isn't already
+// present. It is the upgrade-path safety valve for schema additions:
+// CREATE TABLE IF NOT EXISTS is a no-op when the table already exists, so
+// columns added by newer binaries (e.g. parent_id from the dependent-
+// resources work) never land on databases created by older binaries —
+// which then trip "no such column" when a follow-on CREATE INDEX runs.
+//
+// Skips silently if the table doesn't yet exist (fresh install — the
+// CREATE TABLE migration will create it with the column already declared)
+// or if the column already exists. Runs on the pinned migration
+// connection so it sees the writes performed by the in-flight BEGIN
+// IMMEDIATE transaction; using s.db here would route through the pool
+// and BUSY against the holding writer under concurrent migrators.
+func (s *Store) ensureColumn(ctx context.Context, conn *sql.Conn, table, column, decl string) error {
+ var name string
+ err := conn.QueryRowContext(ctx,
+ `SELECT name FROM sqlite_master WHERE type='table' AND name=?`, table,
+ ).Scan(&name)
+ if err == sql.ErrNoRows {
+ return nil
+ }
+ if err != nil {
+ return fmt.Errorf("checking table %s: %w", table, err)
+ }
+
+ rows, err := conn.QueryContext(ctx, fmt.Sprintf(`PRAGMA table_info("%s")`, table))
+ if err != nil {
+ return fmt.Errorf("table_info %s: %w", table, err)
+ }
+ defer rows.Close()
+ for rows.Next() {
+ var cid int
+ var n, typ string
+ var notnull, pk int
+ var dflt sql.NullString
+ if err := rows.Scan(&cid, &n, &typ, ¬null, &dflt, &pk); err != nil {
+ return fmt.Errorf("scan table_info %s: %w", table, err)
+ }
+ if n == column {
+ return nil
+ }
+ }
+ if err := rows.Err(); err != nil {
+ return fmt.Errorf("iterating table_info %s: %w", table, err)
+ }
+
+ if _, err := conn.ExecContext(ctx, fmt.Sprintf(`ALTER TABLE "%s" ADD COLUMN "%s" %s`, table, column, decl)); err != nil {
+ // A concurrent Open() may have added the column between our
+ // PRAGMA check and this ALTER. SQLite returns SQLITE_ERROR with
+ // "duplicate column name", which busy_timeout does not retry.
+ // The DB is now in the desired state regardless of who won.
+ if strings.Contains(err.Error(), "duplicate column name") {
+ return nil
+ }
+ return fmt.Errorf("add column %s.%s: %w", table, column, err)
+ }
+ return nil
+}
+
+// backfillColumns adds columns that newer binaries declare but that
+// pre-existing databases (created before those columns were added) lack.
+// Must run before the migrations slice so that subsequent CREATE INDEX
+// statements referencing the column can succeed against the upgraded
+// table. Idempotent: safe to call on fresh DBs (table-not-found short-
+// circuit) and on already-current DBs (column-exists short-circuit).
+//
+// Table names are emitted bare (no safeName) — ensureColumn double-quotes
+// them at SQL emit time and uses parameter binding for the sqlite_master
+// lookup, so the values flow as Go string literals first and SQL
+// identifiers second. Wrapping with safeName here would embed literal
+// double-quote characters into the Go string and break compilation for
+// any spec whose dependent-resource snake_cased name is a SQL reserved
+// word.
+func (s *Store) backfillColumns(ctx context.Context, conn *sql.Conn) error {
+ for _, c := range []struct{ table, column, decl string }{
+ {table: "leagues", column: "parent_id", decl: "TEXT"},
+ {table: "sync_state", column: "last_cursor", decl: "TEXT"},
+ {table: "sync_state", column: "last_synced_at", decl: "DATETIME"},
+ {table: "sync_state", column: "total_count", decl: "INTEGER DEFAULT 0"},
+ } {
+ if err := s.ensureColumn(ctx, conn, c.table, c.column, c.decl); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+func (s *Store) migrate(ctx context.Context) error {
+ conn, err := s.db.Conn(ctx)
+ if err != nil {
+ return fmt.Errorf("acquiring migration connection: %w", err)
+ }
+ defer conn.Close()
+
+ // Read user_version before the migration lock so an old binary
+ // opening a newer-schema DB rejects immediately. WAL readers don't
+ // normally block on writers, but the fresh-DB WAL-init race can BUSY
+ // a SELECT — share the lock's deadline so total budget stays bounded.
+ deadline := time.Now().Add(migrationLockTimeout)
+ var current int
+ if err := retryOnBusy(ctx, deadline, "reading schema version", func() error {
+ return conn.QueryRowContext(ctx, `PRAGMA user_version`).Scan(¤t)
+ }); err != nil {
+ return err
+ }
+ if current > StoreSchemaVersion {
+ return fmt.Errorf("database schema version %d is newer than supported version %d; upgrade the CLI binary or open an older database", current, StoreSchemaVersion)
+ }
+
+ migrations := []string{
+ `CREATE TABLE IF NOT EXISTS resources (
+ id TEXT NOT NULL,
+ resource_type TEXT NOT NULL,
+ data JSON NOT NULL,
+ synced_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+ updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+ PRIMARY KEY (resource_type, id)
+ )`,
+ `CREATE INDEX IF NOT EXISTS idx_resources_type ON resources(resource_type)`,
+ `CREATE INDEX IF NOT EXISTS idx_resources_synced ON resources(synced_at)`,
+ `CREATE TABLE IF NOT EXISTS sync_state (
+ resource_type TEXT PRIMARY KEY,
+ last_cursor TEXT,
+ last_synced_at DATETIME,
+ total_count INTEGER DEFAULT 0
+ )`,
+ resourcesFTSCreateSQL,
+ `CREATE TABLE IF NOT EXISTS leagues (
+ id TEXT PRIMARY KEY,
+ data JSON NOT NULL,
+ synced_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+ parent_id TEXT
+ )`,
+ `CREATE INDEX IF NOT EXISTS idx_leagues_parent_id ON leagues(parent_id)`,
+ }
+
+ // Run every migration — including the column backfill and the
+ // schema-version stamp — inside a single BEGIN IMMEDIATE transaction
+ // pinned to one connection. IMMEDIATE acquires SQLite's RESERVED lock
+ // at BEGIN time so concurrent migrators serialize on it instead of
+ // racing per-statement and tripping SQLITE_BUSY despite busy_timeout.
+ // modernc.org/sqlite's busy_timeout does not always cover write-write
+ // contention at BEGIN/COMMIT time, so we retry both explicitly on
+ // SQLITE_BUSY for up to migrationLockTimeout.
+ return withMigrationLock(ctx, conn, deadline, func() error {
+ // Re-read user_version inside the lock. This is load-bearing,
+ // not paranoid: between the pre-lock read above and our
+ // successful BEGIN IMMEDIATE, a newer-binary peer may have
+ // committed a higher version stamp. Without this re-read, an
+ // older binary (smaller StoreSchemaVersion) would proceed to
+ // stamp its own lower version at the end of the closure,
+ // silently downgrading user_version on a schema that's already
+ // at the newer level. Future maintainers: leave this read in.
+ var current int
+ if err := conn.QueryRowContext(ctx, `PRAGMA user_version`).Scan(¤t); err != nil {
+ return fmt.Errorf("reading schema version: %w", err)
+ }
+ if current > StoreSchemaVersion {
+ return fmt.Errorf("database schema version %d is newer than supported version %d; upgrade the CLI binary or open an older database", current, StoreSchemaVersion)
+ }
+
+ if current < 2 {
+ if err := s.migrateResourcesCompositeKey(ctx, conn); err != nil {
+ return fmt.Errorf("migrating resources composite key: %w", err)
+ }
+ }
+
+ if err := s.backfillColumns(ctx, conn); err != nil {
+ return fmt.Errorf("backfilling columns: %w", err)
+ }
+ for _, m := range migrations {
+ if _, err := conn.ExecContext(ctx, m); err != nil {
+ return fmt.Errorf("migration failed: %w", err)
+ }
+ }
+ // Stamp the schema version. On a fresh DB this writes the current
+ // StoreSchemaVersion; on an already-stamped DB this is a no-op
+ // write of the same value.
+ // An older DB with user_version = 0 and pre-existing tables gets
+ // stamped here after any version-gated rewrites and idempotent
+ // CREATE TABLE IF NOT EXISTS statements have completed.
+ if _, err := conn.ExecContext(ctx, fmt.Sprintf(`PRAGMA user_version = %d`, StoreSchemaVersion)); err != nil {
+ return fmt.Errorf("stamp user_version: %w", err)
+ }
+ return nil
+ })
+}
+
+func (s *Store) migrateResourcesCompositeKey(ctx context.Context, conn *sql.Conn) error {
+ exists, err := tableExists(ctx, conn, "resources")
+ if err != nil {
+ return err
+ }
+ if !exists {
+ return nil
+ }
+
+ composite, err := resourcesTableHasCompositeKey(ctx, conn)
+ if err != nil {
+ return err
+ }
+ if !composite {
+ if _, err := conn.ExecContext(ctx, `CREATE TABLE resources_v2 (
+ id TEXT NOT NULL,
+ resource_type TEXT NOT NULL,
+ data JSON NOT NULL,
+ synced_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+ updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
+ PRIMARY KEY (resource_type, id)
+ )`); err != nil {
+ return fmt.Errorf("creating resources_v2: %w", err)
+ }
+ if _, err := conn.ExecContext(ctx, `INSERT INTO resources_v2 (id, resource_type, data, synced_at, updated_at)
+ SELECT id, resource_type, data, synced_at, updated_at FROM resources`); err != nil {
+ return fmt.Errorf("copying resources rows: %w", err)
+ }
+ if _, err := conn.ExecContext(ctx, `DROP TABLE resources`); err != nil {
+ return fmt.Errorf("dropping old resources table: %w", err)
+ }
+ if _, err := conn.ExecContext(ctx, `ALTER TABLE resources_v2 RENAME TO resources`); err != nil {
+ return fmt.Errorf("renaming resources_v2: %w", err)
+ }
+ }
+
+ // Always rebuild FTS during the v2 transition. The resources table may
+ // already have the composite key, but v1 FTS rowids were scoped by id
+ // alone and must be replaced with resource_type + id rowids.
+ if _, err := conn.ExecContext(ctx, `DROP TABLE IF EXISTS resources_fts`); err != nil {
+ return fmt.Errorf("dropping resources_fts: %w", err)
+ }
+ if _, err := conn.ExecContext(ctx, resourcesFTSCreateSQL); err != nil {
+ return fmt.Errorf("creating resources_fts: %w", err)
+ }
+ if err := rebuildResourcesFTS(ctx, conn); err != nil {
+ return fmt.Errorf("rebuilding resources_fts: %w", err)
+ }
+ return nil
+}
+
+func tableExists(ctx context.Context, conn *sql.Conn, name string) (bool, error) {
+ var count int
+ if err := conn.QueryRowContext(ctx, `SELECT COUNT(*) FROM sqlite_master WHERE type = 'table' AND name = ?`, name).Scan(&count); err != nil {
+ return false, fmt.Errorf("checking table %s: %w", name, err)
+ }
+ return count > 0, nil
+}
+
+func resourcesTableHasCompositeKey(ctx context.Context, conn *sql.Conn) (bool, error) {
+ rows, err := conn.QueryContext(ctx, `PRAGMA table_info(resources)`)
+ if err != nil {
+ return false, fmt.Errorf("reading resources table info: %w", err)
+ }
+ defer rows.Close()
+
+ pk := map[string]int{}
+ for rows.Next() {
+ var cid int
+ var name, typ string
+ var notnull, pkOrder int
+ var dflt sql.NullString
+ if err := rows.Scan(&cid, &name, &typ, ¬null, &dflt, &pkOrder); err != nil {
+ return false, fmt.Errorf("scanning resources table info: %w", err)
+ }
+ pk[name] = pkOrder
+ }
+ if err := rows.Err(); err != nil {
+ return false, fmt.Errorf("reading resources table info rows: %w", err)
+ }
+ return pk["resource_type"] == 1 && pk["id"] == 2, nil
+}
+
+func rebuildResourcesFTS(ctx context.Context, conn *sql.Conn) error {
+ rows, err := conn.QueryContext(ctx, `SELECT id, resource_type, data FROM resources`)
+ if err != nil {
+ return fmt.Errorf("querying resources: %w", err)
+ }
+
+ type resourceRow struct {
+ id string
+ resourceType string
+ data string
+ }
+ var resources []resourceRow
+ for rows.Next() {
+ var r resourceRow
+ if err := rows.Scan(&r.id, &r.resourceType, &r.data); err != nil {
+ rows.Close()
+ return fmt.Errorf("scanning resource: %w", err)
+ }
+ resources = append(resources, r)
+ }
+ if err := rows.Err(); err != nil {
+ rows.Close()
+ return fmt.Errorf("reading resource rows: %w", err)
+ }
+ if err := rows.Close(); err != nil {
+ return fmt.Errorf("closing resource rows: %w", err)
+ }
+
+ for _, r := range resources {
+ if _, err := conn.ExecContext(ctx,
+ `INSERT INTO resources_fts (rowid, id, resource_type, content) VALUES (?, ?, ?, ?)`,
+ ftsRowID(r.resourceType, r.id), r.id, r.resourceType, r.data,
+ ); err != nil {
+ return fmt.Errorf("indexing resource %s/%s: %w", r.resourceType, r.id, err)
+ }
+ }
+ return nil
+}
+
+const (
+ migrationLockTimeout = 30 * time.Second
+ migrationLockBackoffMin = 5 * time.Millisecond
+ migrationLockBackoffMax = 100 * time.Millisecond
+)
+
+// withMigrationLock runs fn inside a BEGIN IMMEDIATE / COMMIT pair on
+// conn, retrying both BEGIN and COMMIT on SQLITE_BUSY against the
+// caller-provided deadline. Sharing the deadline with the pre-lock
+// version read keeps total Open() latency bounded by a single budget.
+// The real upper bound is deadline + one trailing backoff interval
+// (≤100ms) + the driver's busy_timeout for the in-flight Exec, since
+// the deadline is checked after each failed attempt rather than as a
+// hard wall-clock cutoff. fn must use conn (not s.db) so its writes
+// participate in the held transaction.
+func withMigrationLock(ctx context.Context, conn *sql.Conn, deadline time.Time, fn func() error) error {
+ if err := execWithBusyRetry(ctx, conn, "BEGIN IMMEDIATE", "begin migration transaction", deadline); err != nil {
+ return err
+ }
+ committed := false
+ defer func() {
+ if committed {
+ return
+ }
+ // ROLLBACK uses context.Background() so caller-context cancellation
+ // can't strand the connection in an open transaction. A failed
+ // rollback is rare on local SQLite (broken file handle, fatal
+ // driver error) but worth surfacing — silent swallow leaves a
+ // pinned connection returned to the pool with state that will
+ // confuse later queries.
+ if _, rerr := conn.ExecContext(context.Background(), "ROLLBACK"); rerr != nil {
+ fmt.Fprintf(os.Stderr, "warning: store migration rollback failed: %v\n", rerr)
+ }
+ }()
+
+ if err := fn(); err != nil {
+ return err
+ }
+
+ if err := execWithBusyRetry(ctx, conn, "COMMIT", "commit migration transaction", deadline); err != nil {
+ return err
+ }
+ committed = true
+ return nil
+}
+
+// execWithBusyRetry runs stmt on conn and retries on SQLITE_BUSY until
+// deadline. It covers BEGIN IMMEDIATE and COMMIT contention;
+// modernc.org/sqlite's busy_timeout does not reliably cover either when
+// multiple connections race for the WAL write lock.
+func execWithBusyRetry(ctx context.Context, conn *sql.Conn, stmt, label string, deadline time.Time) error {
+ return retryOnBusy(ctx, deadline, label, func() error {
+ _, err := conn.ExecContext(ctx, stmt)
+ return err
+ })
+}
+
+// retryOnBusy runs op and retries it on SQLITE_BUSY/LOCKED until
+// deadline. The same retry shape covers Exec, Query, and any other
+// SQLite call that can race the WAL writer lock — including the
+// pre-lock user_version read, where the WAL initialization race on a
+// fresh DB can BUSY a SELECT that should otherwise succeed under WAL
+// reader/writer concurrency.
+func retryOnBusy(ctx context.Context, deadline time.Time, label string, op func() error) error {
+ backoff := migrationLockBackoffMin
+ for {
+ err := op()
+ if err == nil {
+ return nil
+ }
+ if !isSQLiteBusy(err) {
+ return fmt.Errorf("%s: %w", label, err)
+ }
+ if time.Now().After(deadline) {
+ // The label carries the operation context (e.g. "begin
+ // migration transaction", "reading schema version") — we
+ // don't hardcode "waiting for write lock" because pre-lock
+ // reads also flow through this helper.
+ return fmt.Errorf("%s: timed out after %s under SQLite contention: %w", label, migrationLockTimeout, err)
+ }
+ select {
+ case <-ctx.Done():
+ return fmt.Errorf("%s: %w", label, ctx.Err())
+ case <-time.After(backoff):
+ }
+ backoff = min(backoff*2, migrationLockBackoffMax)
+ }
+}
+
+// isSQLiteBusy reports whether err is a retryable SQLite lock condition.
+// Covers both the file-level WAL writer race (SQLITE_BUSY / "database is
+// locked") and the table-level shared-cache contention (SQLITE_LOCKED /
+// "database table is locked"). The match is on the error string because
+// modernc.org/sqlite does not export an error type the generated code
+// can switch on without dragging the driver package into every store
+// consumer.
+func isSQLiteBusy(err error) bool {
+ if err == nil {
+ return false
+ }
+ msg := err.Error()
+ return strings.Contains(msg, "SQLITE_BUSY") ||
+ strings.Contains(msg, "SQLITE_LOCKED") ||
+ strings.Contains(msg, "database is locked") ||
+ strings.Contains(msg, "database table is locked")
+}
+
+func (s *Store) upsertGenericResourceTx(tx *sql.Tx, resourceType, id string, data json.RawMessage) error {
+ _, err := tx.Exec(
+ `INSERT INTO resources (id, resource_type, data, synced_at, updated_at)
+ VALUES (?, ?, ?, ?, ?)
+ ON CONFLICT(resource_type, id) DO UPDATE SET data = excluded.data, synced_at = excluded.synced_at, updated_at = excluded.updated_at`,
+ id, resourceType, string(data), time.Now(), time.Now(),
+ )
+ if err != nil {
+ return err
+ }
+
+ ftsRowid := ftsRowID(resourceType, id)
+ // Use explicit rowid for FTS5 compatibility with modernc.org/sqlite.
+ // Standard DELETE WHERE column=? may not work on FTS5 virtual tables.
+ if _, err = tx.Exec(`DELETE FROM resources_fts WHERE rowid = ?`, ftsRowid); err != nil {
+ fmt.Fprintf(os.Stderr, "warning: FTS index cleanup failed: %v\n", err)
+ }
+
+ if _, err = tx.Exec(
+ `INSERT INTO resources_fts (rowid, id, resource_type, content)
+ VALUES (?, ?, ?, ?)`,
+ ftsRowid, id, resourceType, string(data),
+ ); err != nil {
+ // FTS insert failure is non-fatal
+ fmt.Fprintf(os.Stderr, "warning: FTS index update failed: %v\n", err)
+ }
+
+ return nil
+}
+
+func (s *Store) Upsert(resourceType, id string, data json.RawMessage) error {
+ s.writeMu.Lock()
+ defer s.writeMu.Unlock()
+ tx, err := s.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+
+ if err := s.upsertGenericResourceTx(tx, resourceType, id, data); err != nil {
+ return err
+ }
+
+ return tx.Commit()
+}
+
+// Propagates sql.ErrNoRows on a miss so callers can distinguish absence from
+// other scan errors via errors.Is.
+func (s *Store) Get(resourceType, id string) (json.RawMessage, error) {
+ var data string
+ err := s.db.QueryRow(
+ `SELECT data FROM resources WHERE resource_type = ? AND id = ?`,
+ resourceType, id,
+ ).Scan(&data)
+ if err != nil {
+ return nil, err
+ }
+ return json.RawMessage(data), nil
+}
+
+func (s *Store) List(resourceType string, limit int) ([]json.RawMessage, error) {
+ if limit <= 0 {
+ limit = 200
+ }
+ rows, err := s.db.Query(
+ `SELECT data FROM resources WHERE resource_type = ? ORDER BY updated_at DESC LIMIT ?`,
+ resourceType, limit,
+ )
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+
+ var results []json.RawMessage
+ for rows.Next() {
+ var data string
+ if err := rows.Scan(&data); err != nil {
+ return nil, err
+ }
+ results = append(results, json.RawMessage(data))
+ }
+ return results, rows.Err()
+}
+
+func (s *Store) Search(query string, limit int) ([]json.RawMessage, error) {
+ if limit <= 0 {
+ limit = 50
+ }
+ rows, err := s.db.Query(
+ `SELECT r.data FROM resources r
+ JOIN resources_fts f ON r.id = f.id AND r.resource_type = f.resource_type
+ WHERE resources_fts MATCH ?
+ ORDER BY rank
+ LIMIT ?`,
+ query, limit,
+ )
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+
+ var results []json.RawMessage
+ for rows.Next() {
+ var data string
+ if err := rows.Scan(&data); err != nil {
+ return nil, err
+ }
+ results = append(results, json.RawMessage(data))
+ }
+ return results, rows.Err()
+}
+
+func extractObjectID(obj map[string]any) string {
+ for _, key := range []string{"id", "ID", "uuid", "slug", "name"} {
+ if v, ok := obj[key]; ok {
+ return fmt.Sprintf("%v", v)
+ }
+ }
+ return ""
+}
+
+// ftsRowID derives a deterministic rowid from a string ID for use with FTS5.
+// modernc.org/sqlite's FTS5 implementation may not support DELETE WHERE column=?
+// on virtual tables, so we use explicit rowids and DELETE WHERE rowid=? instead.
+func ftsRowID(scope, id string) int64 {
+ var h uint64
+ for _, c := range scope {
+ h = h*31 + uint64(c)
+ }
+ h *= 31
+ for _, c := range id {
+ h = h*31 + uint64(c)
+ }
+ return int64(h & 0x7FFFFFFFFFFFFFFF) // ensure positive
+}
+
+// LookupFieldValue resolves a field value from a JSON object map, trying
+// the snake_case key first and the camelCase rendering second. Exported so
+// the sync command's extractID and the upsert path resolve fields the same
+// way — a divergence here produces silent drops on heterogeneous payloads.
+func LookupFieldValue(obj map[string]any, snakeKey string) any {
+ if v, ok := obj[snakeKey]; ok {
+ return sqliteFieldValue(v)
+ }
+ parts := strings.Split(snakeKey, "_")
+ for i := 1; i < len(parts); i++ {
+ if parts[i] == "" {
+ continue
+ }
+ parts[i] = strings.ToUpper(parts[i][:1]) + parts[i][1:]
+ }
+ if v, ok := obj[strings.Join(parts, "")]; ok {
+ return sqliteFieldValue(v)
+ }
+ return nil
+}
+
+func sqliteFieldValue(v any) any {
+ switch v.(type) {
+ case nil, string, bool, int, int64, float64, []byte:
+ return v
+ default:
+ data, err := json.Marshal(v)
+ if err != nil {
+ return fmt.Sprint(v)
+ }
+ return string(data)
+ }
+}
+
+// lookupFieldValue is kept as an unexported alias for in-package callers so
+// the existing UpsertBatch code reads naturally without prefixing every call
+// with the package name.
+func lookupFieldValue(obj map[string]any, snakeKey string) any {
+ return LookupFieldValue(obj, snakeKey)
+}
+// upsertLeaguesTx writes the typed-table portion of a leagues upsert
+// inside an existing transaction. The caller is responsible for the generic
+// resources insert (via upsertGenericResourceTx) and for committing the tx.
+// Splitting this out lets UpsertBatch dispatch typed inserts per item without
+// opening a per-item transaction.
+func (s *Store) upsertLeaguesTx(tx *sql.Tx, id string, obj map[string]any, data json.RawMessage) error {
+ if _, err := tx.Exec(
+ `INSERT INTO leagues (id, data, synced_at, parent_id)
+ VALUES (?, ?, ?, ?)
+ ON CONFLICT(id) DO UPDATE SET data = excluded.data, synced_at = excluded.synced_at, parent_id = excluded.parent_id`,
+ id,
+ string(data),
+ time.Now(),
+ lookupFieldValue(obj, "parent_id"),
+ ); err != nil {
+ return fmt.Errorf("insert into leagues: %w", err)
+ }
+
+ return nil
+}
+
+// UpsertLeagues inserts or updates a leagues record with domain-specific columns.
+func (s *Store) UpsertLeagues(data json.RawMessage) error {
+ var obj map[string]any
+ if err := json.Unmarshal(data, &obj); err != nil {
+ return fmt.Errorf("unmarshaling leagues: %w", err)
+ }
+
+ id := extractObjectID(obj)
+ if id == "" {
+ return fmt.Errorf("missing id for leagues")
+ }
+
+ s.writeMu.Lock()
+ defer s.writeMu.Unlock()
+ tx, err := s.db.Begin()
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+
+ if err := s.upsertGenericResourceTx(tx, "leagues", id, data); err != nil {
+ return err
+ }
+ if err := s.upsertLeaguesTx(tx, id, obj, data); err != nil {
+ return err
+ }
+
+ return tx.Commit()
+}
+
+// resourceIDFieldOverrides projects per-resource IDField (set by the profiler
+// from x-resource-id or response-schema fallback) into a runtime lookup map.
+// UpsertBatch consults this first so the templated path wins over the
+// generic fallback list. Empty when no resource declared an override; the
+// runtime fallback list still applies.
+//
+// Includes both flat resources and dependent (parent-child) resources so a
+// child path-item annotated with x-resource-id resolves the same as a flat
+// path-item.
+var resourceIDFieldOverrides = map[string]string{
+ "games": "game_key",
+}
+
+// genericIDFieldFallbacks is the runtime safety net for resources that did
+// NOT receive a templated IDField. API-specific names belong in spec
+// annotations (x-resource-id), not this list.
+var genericIDFieldFallbacks = []string{"id", "ID", "name", "uuid", "slug", "key", "code", "uid"}
+
+// UpsertBatch inserts or replaces multiple records in a single transaction
+// and returns (stored, extractFailures, err). stored counts rows actually
+// landed; extractFailures counts items that survived JSON unmarshal but had
+// no extractable primary key (templated IDField AND generic fallback both
+// missed). callers (sync.go.tmpl) compare these against len(items) to emit
+// the per-item primary_key_unresolved warning and the F4b
+// stored_count_zero_after_extraction probe.
+//
+// For resource types that have a domain-specific typed table, the per-item
+// generic insert is followed by a dispatch to the matching upsert<Pascal>Tx
+// inside the same transaction. Without that dispatch, paginated syncs would
+// only populate the generic resources table — typed tables (and indexed
+// columns like parent_id added by dependent-resource sync) would stay empty.
+func (s *Store) UpsertBatch(resourceType string, items []json.RawMessage) (int, int, error) {
+ s.writeMu.Lock()
+ defer s.writeMu.Unlock()
+ tx, err := s.db.Begin()
+ if err != nil {
+ return 0, 0, fmt.Errorf("starting batch transaction: %w", err)
+ }
+ defer tx.Rollback()
+
+ var stored, skippedCount, extractFailures int
+ for _, item := range items {
+ var obj map[string]any
+ if err := json.Unmarshal(item, &obj); err != nil {
+ skippedCount++
+ continue
+ }
+ // Templated IDField wins; generic fallback list runs second when
+ // the override is empty OR the override field is absent on this
+ // particular item (response shape mismatches happen even when the
+ // spec declares x-resource-id).
+ var id string
+ if override, ok := resourceIDFieldOverrides[resourceType]; ok && override != "" {
+ if v := lookupFieldValue(obj, override); v != nil {
+ s := fmt.Sprintf("%v", v)
+ if s != "" && s != "<nil>" {
+ id = s
+ }
+ }
+ }
+ if id == "" {
+ for _, key := range genericIDFieldFallbacks {
+ if v := lookupFieldValue(obj, key); v != nil {
+ s := fmt.Sprintf("%v", v)
+ if s != "" && s != "<nil>" {
+ id = s
+ break
+ }
+ }
+ }
+ }
+ if id == "" {
+ skippedCount++
+ extractFailures++
+ continue
+ }
+
+ if err := s.upsertGenericResourceTx(tx, resourceType, id, item); err != nil {
+ return 0, extractFailures, fmt.Errorf("upserting %s/%s: %w", resourceType, id, err)
+ }
+
+ switch resourceType {
+ case "leagues":
+ if err := s.upsertLeaguesTx(tx, id, obj, item); err != nil {
+ return 0, extractFailures, fmt.Errorf("typed upsert for %s/%s: %w", resourceType, id, err)
+ }
+ }
+ stored++
+ }
+
+ // Warn when most items in a batch lack an extractable ID — this likely
+ // means the API uses a primary key field we don't recognize yet.
+ if skippedCount > 0 && len(items) > 0 && skippedCount*2 > len(items) {
+ fmt.Fprintf(os.Stderr, "warning: %d/%d %s items skipped (no extractable ID field found)\n", skippedCount, len(items), resourceType)
+ }
+
+ if err := tx.Commit(); err != nil {
+ return 0, extractFailures, err
+ }
+ return stored, extractFailures, nil
+}
+
+func (s *Store) SaveSyncState(resourceType, cursor string, count int) error {
+ s.writeMu.Lock()
+ defer s.writeMu.Unlock()
+ _, err := s.db.Exec(
+ `INSERT INTO sync_state (resource_type, last_cursor, last_synced_at, total_count)
+ VALUES (?, ?, ?, ?)
+ ON CONFLICT(resource_type) DO UPDATE SET last_cursor = excluded.last_cursor,
+ last_synced_at = excluded.last_synced_at, total_count = excluded.total_count`,
+ resourceType, cursor, time.Now(), count,
+ )
+ return err
+}
+
+func (s *Store) GetSyncState(resourceType string) (cursor string, lastSynced time.Time, count int, err error) {
+ err = s.db.QueryRow(
+ `SELECT last_cursor, last_synced_at, total_count FROM sync_state WHERE resource_type = ?`,
+ resourceType,
+ ).Scan(&cursor, &lastSynced, &count)
+ if err == sql.ErrNoRows {
+ return "", time.Time{}, 0, nil
+ }
+ return
+}
+
+// SaveSyncCursor stores the pagination cursor for a resource type.
+func (s *Store) SaveSyncCursor(resourceType, cursor string) error {
+ s.writeMu.Lock()
+ defer s.writeMu.Unlock()
+ _, err := s.db.Exec(
+ `INSERT INTO sync_state (resource_type, last_cursor, last_synced_at, total_count)
+ VALUES (?, ?, CURRENT_TIMESTAMP, 0)
+ ON CONFLICT(resource_type) DO UPDATE SET last_cursor = ?, last_synced_at = CURRENT_TIMESTAMP`,
+ resourceType, cursor, cursor,
+ )
+ return err
+}
+
+// GetSyncCursor returns the last pagination cursor for a resource type.
+func (s *Store) GetSyncCursor(resourceType string) string {
+ var cursor sql.NullString
+ s.db.QueryRow("SELECT last_cursor FROM sync_state WHERE resource_type = ?", resourceType).Scan(&cursor)
+ if cursor.Valid {
+ return cursor.String
+ }
+ return ""
+}
+
+// ListIDs returns all IDs from a resource's domain table, or from the generic
+// resources table if no domain table exists. Used by dependent sync to iterate parents.
+//
+// resourceType is never interpolated into SQL directly. We resolve it to a real
+// table name via a parameterized sqlite_master lookup; only that trusted name is
+// substituted (double-quoted) into the SELECT. Callers may pass any string.
+func (s *Store) ListIDs(resourceType string) ([]string, error) {
+ var table string
+ err := s.db.QueryRow(
+ `SELECT name FROM sqlite_master WHERE type='table' AND name=?`,
+ resourceType,
+ ).Scan(&table)
+ var rows *sql.Rows
+ if err == nil && table != "" {
+ rows, err = s.db.Query(fmt.Sprintf(`SELECT id FROM "%s"`, strings.ReplaceAll(table, `"`, `""`)))
+ }
+ if err != nil || table == "" {
+ // Fall back to generic resources table
+ rows, err = s.db.Query("SELECT id FROM resources WHERE resource_type = ?", resourceType)
+ if err != nil {
+ return nil, err
+ }
+ }
+ defer rows.Close()
+
+ var ids []string
+ for rows.Next() {
+ var id string
+ if err := rows.Scan(&id); err != nil {
+ continue
+ }
+ ids = append(ids, id)
+ }
+ return ids, rows.Err()
+}
+
+// ListField returns values of a named field from a resource's domain table,
+// or from the generic resources table via json_extract when no typed column
+// exists. Used by dependent sync to iterate parents when a spec-declared
+// walker extracts a non-PK field (Endpoint.Walker.KeyField in the upstream
+// printing-press repo) for the child path's placeholder.
+//
+// Defense in depth: field is validated against validIdentifierRE at entry
+// — the regex pins it to SQL-safe identifier shape covering both the
+// typed-column primary path AND the json_extract fallback (where
+// pragma_table_info validation would never run if the parent's domain
+// table doesn't exist yet). resourceType is never interpolated into SQL
+// directly; we resolve it to a real table name via a parameterized
+// sqlite_master lookup. Only validated names are substituted
+// (double-quoted) into the SELECT. Mirrors ListIDs's defense pattern so
+// callers may pass any string.
+func (s *Store) ListField(resourceType, field string) ([]string, error) {
+ if !validIdentifierRE.MatchString(field) {
+ return nil, fmt.Errorf("ListField: invalid field name %q (must match %s)", field, validIdentifierRE.String())
+ }
+ var table string
+ err := s.db.QueryRow(
+ `SELECT name FROM sqlite_master WHERE type='table' AND name=?`,
+ resourceType,
+ ).Scan(&table)
+ var rows *sql.Rows
+ if err == nil && table != "" {
+ // Validate the column exists on the resolved table before splicing
+ // it into the SELECT. pragma_table_info is parameterizable.
+ var colName string
+ colErr := s.db.QueryRow(
+ `SELECT name FROM pragma_table_info(?) WHERE name=?`,
+ table, field,
+ ).Scan(&colName)
+ if colErr == nil && colName != "" {
+ qTable := strings.ReplaceAll(table, `"`, `""`)
+ qCol := strings.ReplaceAll(colName, `"`, `""`)
+ // DISTINCT: callers iterate the returned values as parent keys
+ // for child-resource fan-out. Multiple parent rows sharing a
+ // key_field value (legal for non-PK fields) would otherwise
+ // cause the child endpoint to be fetched once per duplicate row.
+ rows, err = s.db.Query(fmt.Sprintf(
+ `SELECT DISTINCT "%s" FROM "%s" WHERE "%s" IS NOT NULL AND "%s" != ''`,
+ qCol, qTable, qCol, qCol,
+ ))
+ } else {
+ err = colErr
+ }
+ }
+ if err != nil || rows == nil {
+ // Fall back to generic resources table via json_extract. Path is
+ // Sprintf'd into the SQL string (matches ResolveByName below).
+ // DISTINCT for the same reason as the typed-column path above.
+ fallback := fmt.Sprintf(
+ `SELECT DISTINCT json_extract(data, '$.%s') FROM resources WHERE resource_type = ? AND json_extract(data, '$.%s') IS NOT NULL`,
+ field, field,
+ )
+ rows, err = s.db.Query(fallback, resourceType)
+ if err != nil {
+ return nil, err
+ }
+ }
+ defer rows.Close()
+
+ var values []string
+ for rows.Next() {
+ var v sql.NullString
+ if err := rows.Scan(&v); err == nil && v.Valid && v.String != "" {
+ values = append(values, v.String)
+ }
+ }
+ return values, rows.Err()
+}
+
+// GetLastSyncedAt returns the last sync timestamp for a resource type.
+func (s *Store) GetLastSyncedAt(resourceType string) string {
+ var ts sql.NullString
+ s.db.QueryRow("SELECT last_synced_at FROM sync_state WHERE resource_type = ?", resourceType).Scan(&ts)
+ if ts.Valid {
+ return ts.String
+ }
+ return ""
+}
+
+// ClearSyncCursors resets all sync state for a full resync.
+func (s *Store) ClearSyncCursors() error {
+ s.writeMu.Lock()
+ defer s.writeMu.Unlock()
+ _, err := s.db.Exec("DELETE FROM sync_state")
+ return err
+}
+
+// Query executes a raw SQL query and returns the rows.
+// Used by workflow commands that need custom queries against the local store.
+func (s *Store) Query(query string, args ...any) (*sql.Rows, error) {
+ return s.db.Query(query, args...)
+}
+
+func (s *Store) Count(resourceType string) (int, error) {
+ var count int
+ err := s.db.QueryRow(
+ `SELECT COUNT(*) FROM resources WHERE resource_type = ?`,
+ resourceType,
+ ).Scan(&count)
+ return count, err
+}
+
+func (s *Store) Status() (map[string]int, error) {
+ rows, err := s.db.Query(
+ `SELECT resource_type, COUNT(*) FROM resources GROUP BY resource_type ORDER BY resource_type`,
+ )
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+
+ status := make(map[string]int)
+ for rows.Next() {
+ var rt string
+ var count int
+ if err := rows.Scan(&rt, &count); err != nil {
+ return nil, err
+ }
+ status[rt] = count
+ }
+ return status, rows.Err()
+}
+
+// ResolveByName resolves a human-readable name to a UUID from synced data.
+// If the input is already a UUID, it is returned as-is.
+// matchFields are JSON field names to search against (e.g., "name", "key", "email").
+func (s *Store) ResolveByName(resourceType string, input string, matchFields ...string) (string, error) {
+ if IsUUID(input) {
+ return input, nil
+ }
+
+ var matches []string
+ for _, field := range matchFields {
+ query := fmt.Sprintf(
+ `SELECT id FROM resources WHERE resource_type = ? AND LOWER(json_extract(data, '$.%s')) = LOWER(?)`,
+ field,
+ )
+ rows, err := s.db.Query(query, resourceType, input)
+ if err != nil {
+ continue
+ }
+ for rows.Next() {
+ var id string
+ if rows.Scan(&id) == nil {
+ // Deduplicate
+ found := false
+ for _, m := range matches {
+ if m == id {
+ found = true
+ break
+ }
+ }
+ if !found {
+ matches = append(matches, id)
+ }
+ }
+ }
+ rows.Close()
+ }
+
+ switch len(matches) {
+ case 0:
+ return "", fmt.Errorf("%s %q not found in local store. Run 'sync' first, or use the UUID directly", resourceType, input)
+ case 1:
+ return matches[0], nil
+ default:
+ hint := matches[0]
+ if len(matches) > 5 {
+ hint = strings.Join(matches[:5], ", ") + "..."
+ } else {
+ hint = strings.Join(matches, ", ")
+ }
+ return "", fmt.Errorf("ambiguous: %q matches %d %s entries (%s). Use the exact UUID instead", input, len(matches), resourceType, hint)
+ }
+}
diff --git a/testdata/golden/fixtures/sync-walker-api.yaml b/testdata/golden/fixtures/sync-walker-api.yaml
new file mode 100644
index 00000000..59b24982
--- /dev/null
+++ b/testdata/golden/fixtures/sync-walker-api.yaml
@@ -0,0 +1,44 @@
+name: sync-walker-golden
+description: |
+ Golden fixture for the Tier 2 spec-declared walker mechanism. Demonstrates
+ the shape used by hierarchical APIs (Yahoo Fantasy, Reddit pre-2024,
+ YouTube Data v3, MLB Stats) where the child endpoint's path placeholder
+ is a field on the parent record that is not the parent's primary key.
+
+ Walker shape exercised: leagues declares `walker.parent=games` with
+ `key_field=game_key` and `key_param=game_key`. The child path
+ /games/{game_key}/leagues uses a placeholder whose stem ("game_key")
+ does not map to a parent resource via auto-detection's _id/_key
+ stripping — without the walker, no dependent would be detected.
+
+ Fall-through behavior (walker without key_field → existing parent-PK
+ flow via db.ListIDs) is covered by every existing flat-list CLI's
+ dependent sync code path; not repeated in this fixture.
+version: 1.0.0
+base_url: https://api.example.com
+auth:
+ type: bearer_token
+ env_vars: [WALKER_GOLDEN_TOKEN]
+resources:
+ games:
+ description: Top-level games resource. The list endpoint populates the
+ generic resources table; rows carry a `game_key` field that the
+ walker's leagues endpoint extracts for child fan-out.
+ endpoints:
+ list:
+ method: GET
+ path: /games
+ description: List games
+ id_field: game_key
+ leagues:
+ description: Leagues, fetched per-game by walking games and extracting
+ each game's game_key into the child path.
+ endpoints:
+ list:
+ method: GET
+ path: /games/{game_key}/leagues
+ description: List leagues for a game
+ walker:
+ parent: games
+ key_field: game_key
+ key_param: game_key
← dc6fe879 fix(cli): preserve unknown-shape records in compactListField
·
back to Cli Printing Press
·
docs(skills): clarify wrapper-only catalog entries have no g 46481718 →