// 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"
	"github.com/spf13/cobra"
	"net/url"
	"os"
	"regexp"
	"strconv"
	"strings"
	"sync"
	"sync-walker-golden-pp-cli/internal/cliutil"
	"sync-walker-golden-pp-cli/internal/store"
	"sync/atomic"
	"time"
)

// 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
	var paramFlags []string
	var resourceParamFlags []string
	var globalParamFlags []string

	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.

Resource scoping:
  --resources runs the named top-level resources, plus any parent-keyed
  dependent whose own name is listed OR whose parent table is listed.
  Naming a parent therefore cascades to its dependents (so "sync this
  parent and its children" works without listing every nested resource
  by hand). There is no flag today to suppress the cascade for a named
  parent. To run a dependent without re-syncing its parent, list only
  the dependent by name; the parent table must already be populated
  from a prior sync.`,
		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 {
			userParams, err := parseSyncUserParams(paramFlags, resourceParamFlags, globalParamFlags)
			if err != nil {
				return usageErr(err)
			}

			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()
			}

			// Reject --resource-param keys that don't match a known resource.
			// Validates against the full top-level + dependent set, not the
			// user-filtered `resources` slice, so legitimate cases like
			// "filter to A, but apply param to B if it gets synced" still
			// catch typos without false positives.
			if err := userParams.validateResourceNames(knownSyncResourceNames()); err != nil {
				return usageErr(err)
			}

			// --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")
				}
			}
			// effectiveLatestOnly drives the max_pages_cap_hit suppression
			// below. It must reflect whether --latest-only is actually the
			// cap source — i.e., only when --since is empty. If --since wins
			// (block above), --latest-only is a no-op for maxPages and any
			// cap hit reflects the default --max-pages 100 limit, which is
			// a real anomaly worth surfacing.
			effectiveLatestOnly := latestOnly && since == ""

			// 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
			}
			// Under PRINTING_PRESS_VERIFY=1 (mock/dry-run), all goroutines
			// reach SQLite without the natural serialization that network
			// latency provides in real syncs, so the worker pool races on
			// the writer and trips SQLITE_BUSY despite _busy_timeout.
			if cliutil.IsVerifyEnv() {
				concurrency = 1
			}

			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, effectiveLatestOnly, userParams)
						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, effectiveLatestOnly, parentFilter, userParams)
			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. Naming a parent also runs its parent-keyed dependents (see Long help for scoping).")
	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).")
	cmd.Flags().StringArrayVar(&paramFlags, "param", nil, "Extra query param to inject into flat-list sync requests (repeatable, key=value). Skipped on path-scoped dependent requests so a top-level scope like workspace=<id> does not double up on /parents/<id>/children calls. Use --global-param to inject everywhere. Avoid pagination keys (limit/since/cursor) — overriding them corrupts resume state.")
	cmd.Flags().StringArrayVar(&resourceParamFlags, "resource-param", nil, "Per-resource extra query param (repeatable, resource:key=value). Wins over --param and --global-param when keys conflict.")
	cmd.Flags().StringArrayVar(&globalParamFlags, "global-param", nil, "Extra query param to inject into every sync request including dependent path-scoped calls (repeatable, key=value). Use when an API requires a scope on every call regardless of path nesting.")

	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.
// channel_workflow.go.tmpl mirrors the trailing dates arg conditional;
// keep both call sites in sync if this signature changes.
func syncResource(c interface {
	Get(string, map[string]string) (json.RawMessage, error)
	RateLimit() float64
}, db *store.Store, resource, sinceTS string, full bool, maxPages int, latestOnly bool, userParams *syncUserParams) 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{}

		if resourceSupportsPagination(resource) {
			params[pageSize.limitParam] = strconv.Itoa(pageSize.limit)
			if cursor != "" {
				params[pageSize.cursorParam] = cursor
			}
		}

		// Set since filter
		if effectiveSince != "" {
			params[sinceParam] = effectiveSince
		}

		// Apply user-supplied --param / --resource-param overrides last so they
		// win over spec-derived defaults (e.g. forcing mine=true on a list
		// endpoint whose OpenAPI spec marks the filter optional).
		userParams.applyTo(resource, params, false)

		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.Fprintln(os.Stdout, syncErrorJSON(resource, "", err))
			}
			return syncResult{Resource: resource, Count: totalCount, Err: fmt.Errorf("fetching %s: %w", resource, err), Duration: time.Since(started)}
		}

		// Dry-run sentinel: client.dryRun returns `{"dry_run": true}` instead
		// of a real response when --dry-run is set. The upsert path below
		// would otherwise fail with "missing id for <resource>" because the
		// sentinel has no items and no id; emit a synthetic success event so
		// validate-narrative --full-examples (which auto-appends --dry-run)
		// sees a clean exit.
		if isDryRunResponse(data) {
			if !humanFriendly {
				fmt.Fprintf(os.Stdout, `{"event":"sync_dryrun","resource":"%s"}`+"\n", resource)
			}
			return syncResult{Resource: resource, Count: 0, 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)

		// Page-int paginator fallback: when the API paginates by integer
		// ?page=N and emits no body cursor, treat a full page as a signal
		// to advance numerically. Without this the loop breaks after page
		// 1 even though more pages exist (the original symptom in #1296).
		// Guard on cursorType, not cursorParam name, so all canonical
		// spellings (page / page_number / pageNumber / page[number]) work.
		if pageSize.cursorType == "page" && nextCursor == "" && len(items) >= pageSize.limit {
			currentPage, _ := strconv.Atoi(cursor)
			if currentPage < 1 {
				currentPage = 1
			}
			nextCursor = strconv.Itoa(currentPage + 1)
			hasMore = true
		}

		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.Fprintln(os.Stdout, syncErrorJSON(resource, "", err))
				}
				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.Fprintln(os.Stdout, syncErrorJSON(resource, "", err))
			}
			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))
			}
		}

		pagesFetched++

		// Enforce page ceiling to prevent runaway syncs on large-catalog APIs.
		// Suppress the cap-hit warning when --latest-only is the cap source:
		// the template pinned maxPages=1 by user intent, and emitting one
		// warning per paginated resource would mask real sync_anomaly /
		// sync_error output in the same stream.
		if maxPages > 0 && pagesFetched >= maxPages {
			if !latestOnly {
				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 !resourceSupportsPagination(resource) {
			break
		}
		if !hasMore || len(items) < pageSize.limit {
			break
		}
		if nextCursor == "" {
			if pageSize.cursorParam == "offset" {
				// Cursor-based APIs return the next cursor in the envelope.
				// Offset-based APIs carry their pagination position client-side.
				currentOffset, _ := strconv.Atoi(cursor)
				nextCursor = strconv.Itoa(currentOffset + pageSize.limit)
			} else {
				// A cursor-based API reporting has_more without a next cursor
				// cannot advance safely; stop instead of looping silently.
				break
			}
		}

		// 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)
		}

		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
	cursorType  string // paginator class: "", "cursor", "page_token", "offset", "page"
	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",
		cursorType:  "",
		limitParam:  "limit",
		limit:       100,
	}
}

func resourceSupportsPagination(resource string) bool {
	switch resource {
	}
	return false
}

// 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
}

// isDryRunResponse detects the `{"dry_run": true}` sentinel that
// client.dryRun returns instead of a real API response. The sync loop
// uses this to short-circuit before the upsert path, which would
// otherwise fail with "missing id for <resource>" against a sentinel
// that has no items and no id. The check requires exactly one key
// (dry_run) so a live API response that happens to include a top-level
// dry_run field alongside real data isn't misclassified as the
// sentinel and silently zero out the sync count.
func isDryRunResponse(data json.RawMessage) bool {
	var envelope map[string]json.RawMessage
	if err := json.Unmarshal(data, &envelope); err != nil {
		return false
	}
	if len(envelope) != 1 {
		return false
	}
	raw, ok := envelope["dry_run"]
	if !ok {
		return false
	}
	var v bool
	return json.Unmarshal(raw, &v) == nil && v
}

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",
	}
}

// knownSyncResourceNames returns every resource name sync will accept —
// flat resources plus any parent-child dependents. Used by --resource-param
// validation to reject misspellings before they become silent no-ops.
func knownSyncResourceNames() []string {
	names := defaultSyncResources()
	for _, dep := range dependentResourceDefs() {
		names = append(names, dep.Name)
	}
	return names
}

// 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, latestOnly bool, parentFilter []string, userParams *syncUserParams) []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, latestOnly, userParams)
		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, latestOnly bool, userParams *syncUserParams) 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{}
			if resourceSupportsPagination(dep.Name) {
				params[pageSize.limitParam] = strconv.Itoa(pageSize.limit)
				if cursor != "" {
					params[pageSize.cursorParam] = cursor
				}
			}
			if depSinceTS != "" {
				params[depSinceParam] = depSinceTS
			}

			// Apply user flags last so they win over spec-derived cursor/since/limit.
			// Dependent path: --param is skipped (already scoped by the parent path
			// segment); --global-param and --resource-param still apply.
			userParams.applyTo(dep.Name, params, true)

			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)
				} else {
					// Non-warning failures were previously silent in JSON mode —
					// operators only saw the missing rows. Emit a structured
					// sync_error so the API body and status are inspectable.
					fmt.Fprintln(os.Stdout, syncErrorJSON(dep.Name, parentID, err))
				}
				break
			}

			items, nextCursor, hasMore := extractPageItems(data, pageSize.cursorParam)

			// Page-int paginator fallback: mirrors syncResource so dependent
			// resources on integer ?page=N APIs also advance past page 1.
			// Guard on cursorType to cover every canonical spelling.
			if pageSize.cursorType == "page" && nextCursor == "" && len(items) >= pageSize.limit {
				currentPage, _ := strconv.Atoi(cursor)
				if currentPage < 1 {
					currentPage = 1
				}
				nextCursor = strconv.Itoa(currentPage + 1)
				hasMore = true
			}

			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 !latestOnly {
					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 !resourceSupportsPagination(dep.Name) {
				break
			}
			if !hasMore || len(items) < pageSize.limit {
				break
			}
			if nextCursor == "" {
				if pageSize.cursorParam == "offset" {
					// Cursor-based APIs return the next cursor in the envelope.
					// Offset-based APIs carry their pagination position client-side.
					currentOffset, _ := strconv.Atoi(cursor)
					nextCursor = strconv.Itoa(currentOffset + pageSize.limit)
				} else {
					// A cursor-based API reporting has_more without a next cursor
					// cannot advance safely; stop instead of looping silently.
					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. Order matters: vendor
// identifier names (gid, sid, uid, uuid, guid) take precedence over `name`
// so APIs like Asana (gid) and Twilio (sid) don't fall through to a display
// field and upsert on names — see #1394.
var genericIDFieldFallbacks = []string{"id", "ID", "gid", "sid", "uid", "uuid", "guid", "name", "slug", "key", "code"}

// pageItemKeys is scanned in priority order; lowercase REST-convention keys
// come first, PascalCase .NET variants second. Without the PascalCase row,
// {"Items": [...]} envelopes fall through to the ambiguity scan and a
// single-array sibling miscount silently truncates sync.
var pageItemKeys = []string{
	"data", "results", "items", "records", "nodes", "entries",
	"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 ""
}
