// Copyright 2026 printing-press-golden and contributors. 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 (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"github.com/spf13/cobra"
	"io"
	"net/url"
	"os"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"tier-routing-golden-pp-cli/internal/client"
	"tier-routing-golden-pp-cli/internal/cliutil"
	"tier-routing-golden-pp-cli/internal/store"
	"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 noPrune 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. This API does
not declare temporal sync filters, so sync performs full pagination unless an
explicit resource declares its own incremental filter.
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
  tier-routing-golden-pp-cli sync

  # Sync specific resources only
  tier-routing-golden-pp-cli sync --resources channels,messages

  # Full resync (ignore previous checkpoint)
  tier-routing-golden-pp-cli sync --full

  # Incremental sync: only records from the last 7 days
  tier-routing-golden-pp-cli sync --since 7d

  # Parallel sync with 8 workers
  tier-routing-golden-pp-cli sync --concurrency 8

  # Latest-only: refresh head of each resource, no historical backfill
  tier-routing-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("tier-routing-golden-pp-cli")
			}

			db, err := store.OpenWithContext(cmd.Context(), dbPath)
			if err != nil {
				return fmt.Errorf("opening local database: %w", err)
			}
			defer db.Close()

			syncEventWriter := cmd.OutOrStdout()

			// 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.
			// Skip under --dry-run: a preview must not mutate sync-state (issue #2935).
			if full && !c.DryRun {
				for _, resource := range resources {
					_ = db.SaveSyncState(resource, "", 0)
				}
			}

			if cliutil.IsDogfoodEnv() && !cmd.Flags().Changed("max-pages") {
				maxPages = 1
			}

			// --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". Skip under --dry-run:
					// a preview must not mutate sync-state (issue #2935).
					if !c.DryRun {
						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 an explicit operator limit or the dogfood
			// safety 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()
			// prune gates deletion reconciliation: a full sync prunes local rows
			// the API no longer returns within a fully-enumerated partition, unless
			// --no-prune disables it. Flat tenant-scoped reconcile (per resource)
			// and dependent per-parent reconcile share this gate.
			prune := full && !noPrune
			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(cmd.Context(), syncClientForResource(c, resource), db, resource, sinceTS, full, maxPages, effectiveLatestOnly, prune, userParams, syncEventWriter)
						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
			var failedResources []string
			var criticalFailedResources []string
			var firstErr error
			var firstPlaceholderErr error
			for res := range results {
				if res.Err != nil {
					if humanFriendly {
						fmt.Fprintf(os.Stderr, "  %s: error: %v\n", res.Resource, res.Err)
					}
					errCount++
					failedResources = append(failedResources, res.Resource)
					if firstErr == nil {
						firstErr = res.Err
					}
					if firstPlaceholderErr == nil && errors.Is(res.Err, client.ErrPlaceholderCredential) {
						firstPlaceholderErr = res.Err
					}
					if criticalResources[res.Resource] {
						criticalErrCount++
						criticalFailedResources = append(criticalFailedResources, res.Resource)
					}
				} 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(syncEventWriter, `{"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 firstPlaceholderErr != nil {
				return classifyAPIError(firstPlaceholderErr, flags)
			}
			if strict && errCount > 0 {
				return errors.New(describeFailedResources(errCount, failedResources))
			}
			if criticalErrCount > 0 {
				return errors.New(describeCriticalFailedResources(criticalErrCount, criticalFailedResources))
			}
			if successCount == 0 {
				if warnCount > 0 && errCount == 0 {
					return fmt.Errorf("%d resource(s) completed with warnings but no successful syncs", warnCount)
				}
				if errCount > 0 {
					return errors.New(describeFailedResources(errCount, failedResources))
				}
			}
			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(syncEventWriter, `{"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().BoolVar(&noPrune, "no-prune", false, "Disable deletion reconciliation on --full (by default a full sync prunes local rows the API no longer returns for a fully-enumerated parent partition)")
	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", "", "SQLite database file path (default: resolved data directory data.db)")
	cmd.Flags().IntVar(&maxPages, "max-pages", 0, "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(ctx context.Context, c interface {
	Get(context.Context, string, map[string]string) (json.RawMessage, error)
	RateLimit() float64
}, db *store.Store, resource, sinceTS string, full bool, maxPages int, latestOnly bool, prune bool, userParams *syncUserParams, syncEvents io.Writer) syncResult {
	started := time.Now()
	if syncEvents == nil {
		syncEvents = io.Discard
	}

	if !humanFriendly {
		fmt.Fprintf(syncEvents, `{"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(syncEvents, "%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)
	if !full {
		if storedCount, err := db.Count(resource); err == nil && storedCount == 0 {
			existingCursor = ""
			lastSynced = time.Time{}
		}
	}

	// 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(syncEvents, `{"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 = ""
	}
	if effectiveSince != "" {
		effectiveSince = formatSyncSinceValue(effectiveSince, syncResourceSinceParamFormat(resource))
	}

	cursor := existingCursor
	pageSize := determinePaginationDefaults(resource)
	var progressCount int64
	pagesFetched := 0
	lastNextCursor := ""
	capExitHit := false
	capExitCursor := ""
	// 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

	// Flat tenant-scoped reconcile bookkeeping (mirrors the dependent loop's
	// partitionOutcome machinery). flatReconcilable gates all of it: only
	// resources classified reconcileMode=="flat" collect seen IDs and prune.
	// outcome.complete is set ONLY at proven natural ends; any abnormal break
	// sets outcome.reason and leaves complete=false so the reconcile SKIPS.
	flatReconcilable := resourceReconcileMode(resource) == "flat"
	outcome := partitionOutcome{}
	var seenIDs []string

	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(ctx, path, params)
		if err != nil {
			if w, ok := isSyncAccessWarning(err); ok {
				if !humanFriendly {
					fmt.Fprintln(syncEvents, syncWarningJSON(resource, "", w.Status, w.Reason, w.Message))
				}
				return syncResult{Resource: resource, Count: totalCount, Warn: fmt.Errorf("skipped due to insufficient access: %s (%s)", resource, w.Reason), Duration: time.Since(started)}
			}
			if !humanFriendly {
				fmt.Fprintln(syncEvents, 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(syncEvents, `{"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, responsePathForResource(resource, path)...)

		// 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 && pageAllowsPageIntFallback(data) {
			currentPage, _ := strconv.Atoi(cursor)
			if currentPage < 1 {
				currentPage = 1
			}
			nextCursor = strconv.Itoa(currentPage + 1)
			hasMore = true
		}
		if pageSize.cursorType == "offset" && nextCursor == "" && len(items) >= pageSize.limit && pageAllowsPageIntFallback(data) {
			currentOffset, _ := strconv.Atoi(cursor)
			nextCursor = strconv.Itoa(currentOffset + pageSize.limit)
			hasMore = true
		}

		if len(items) == 0 && len(data) > 0 && !isJSONResponse(data) {
			// Abnormal: a 200 with a non-JSON body means the page was not a
			// trustworthy enumeration — leave outcome.complete=false so reconcile
			// SKIPS. Reuse the dependent loop's reason string for cross-loop
			// consistency.
			outcome.reason = "non_json_200_body"
			if humanFriendly {
				fmt.Fprintf(os.Stderr, "\nwarning: %s returned a 200 response with a non-JSON body; no rows were stored.\n", resource)
			} else {
				fmt.Fprintf(syncEvents, `{"event":"sync_anomaly","resource":"%s","reason":"non_json_200_body"}`+"\n", resource)
			}
			break
		}

		if len(items) == 0 {
			if isEmptyPageResponse(data, responsePathForResource(resource, path)...) {
				// Natural end: the API legitimately returned an empty page.
				outcome.complete = true
				break
			}
			// Single object response - try to store as-is
			if err := upsertSingleObject(db, resource, data); err != nil {
				if !humanFriendly {
					fmt.Fprintln(syncEvents, syncErrorJSON(resource, "", err))
				}
				return syncResult{Resource: resource, Err: err, Duration: time.Since(started)}
			}
			totalCount++
			// Single-object resources are fully enumerated by definition.
			outcome.complete = true
			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(syncEvents, 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(syncEvents, `{"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(syncEvents, `{"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))
		// Collect seen IDs for flat tenant-scoped reconcile. extractID resolves
		// the primary key the same way UpsertBatch does, so seenIDs match stored
		// row IDs. Only non-empty IDs are tracked; reconcile never deletes a row
		// whose ID was seen this run.
		if flatReconcilable {
			for _, it := range items {
				if o, derr := store.DecodeJSONObject(it); derr == nil {
					if id := extractID(resource, o); id != "" {
						seenIDs = append(seenIDs, id)
					}
				}
			}
		}
		if resourceSupportsPagination(resource) && nextCursor == "" && pageSize.cursorParam != "offset" && len(items) >= pageSize.limit && pageMayHaveMore(data, responsePathForResource(resource, path)...) {
			emitSyncMissingPaginationCursorWarning(syncEvents, humanFriendly, resource, "")
		}

		// 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(syncEvents, `{"event":"sync_progress","resource":"%s","fetched":%d,"rate_rps":%.1f}`+"\n", resource, atomic.LoadInt64(&progressCount), currentRate)
			} else {
				fmt.Fprintf(syncEvents, `{"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 {
			truncatedByCap := resourceSupportsPagination(resource) && hasMore
			truncatedByCap = truncatedByCap && len(items) >= pageSize.limit
			if truncatedByCap {
				capExitCursor = nextCursor
			}
			if truncatedByCap && capExitCursor == "" {
				if pageSize.cursorType == "offset" {
					currentOffset, _ := strconv.Atoi(cursor)
					capExitCursor = strconv.Itoa(currentOffset + pageSize.limit)
				} else {
					truncatedByCap = false
				}
			}
			if truncatedByCap && capExitCursor != cursor {
				if !latestOnly {
					capExitHit = true
					if humanFriendly {
						fmt.Fprintf(os.Stderr, "\n  %s: reached --max-pages limit (%d pages, %d items)\n", resource, maxPages, totalCount)
					} else {
						fmt.Fprintf(syncEvents, `{"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)
					}
				}
			}
			// A cap break ends enumeration before exhausting the partition; mirror
			// the dependent loop and treat it as incomplete (reconcile SKIPS). Safe
			// default: when the cap may have truncated, never prune.
			outcome.reason = "max_pages_cap"
			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 {
			// Abnormal: a non-advancing cursor means we cannot prove the partition
			// was fully enumerated — leave outcome.complete=false (reconcile SKIPS).
			outcome.reason = "stuck_cursor"
			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(syncEvents, `{"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) {
			outcome.complete = true // resource declares no pagination: one page is the whole set
			break
		}
		if !hasMore || len(items) < pageSize.limit {
			outcome.complete = true
			break
		}
		if nextCursor == "" {
			if pageSize.cursorType == "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.
				outcome.reason = "cursor_unavailable"
				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
	}

	// Flat tenant-scoped reconcile: prune local rows the API no longer returns
	// within THIS tenant's partition, gated on a proven-complete sync.
	//   - Unknown tenant (resolveTenantID()=="") ⇒ SKIP, zero deletes. This is
	//     the OPPOSITE of the dependent fan-out fallback (which enumerates
	//     unscoped): a flat delete cannot be safely scoped without a tenant, so
	//     we never delete on unknown.
	//   - Incomplete sync (outcome.complete==false) ⇒ SKIP with the recorded
	//     reason; an abnormal break never proves the partition was enumerated.
	if prune && flatReconcilable {
		def := flatReconcileDef(resource)
		tenantUUID := resolveTenantID()
		if tenantUUID == "" {
			fmt.Fprintf(syncEvents, `{"event":"reconcile_skipped","resource":"%s","reason":"unknown-tenant"}`+"\n", resource)
		} else if outcome.complete {
			deleted, rerr := db.ReconcilePartition(
				resource, "$."+def.BodyField, tenantUUID,
				seenIDs, resource, store.CascadeJunctionsFor(resource),
			)
			if rerr != nil {
				fmt.Fprintf(syncEvents, `{"event":"reconcile_error","resource":"%s","scope":"%s","error":%q}`+"\n", resource, tenantUUID, rerr.Error())
			} else {
				fmt.Fprintf(syncEvents, `{"event":"reconcile","resource":"%s","scope":"%s","deleted":%d}`+"\n", resource, tenantUUID, deleted)
			}
		} else {
			fmt.Fprintf(syncEvents, `{"event":"reconcile_skipped","resource":"%s","scope":"%s","reason":%q}`+"\n", resource, tenantUUID, outcome.reason)
		}
	}

	// Final sync state: clear cursor on natural completion, but preserve the
	// resume cursor when an operator intentionally capped the page budget.
	finalCursor := ""
	if capExitHit {
		finalCursor = capExitCursor
	}
	_ = db.SaveSyncState(resource, finalCursor, 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(syncEvents, `{"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(syncEvents, `{"event":"sync_complete","resource":"%s","total":%d,"duration_ms":%d}`+"\n", resource, totalCount, time.Since(started).Milliseconds())
	}

	if consumedTotal > 0 && totalCount == 0 && extractFailureTotal >= consumedTotal {
		return syncResult{
			Resource: resource,
			Count:    0,
			Warn:     fmt.Errorf("%s consumed %d items but stored 0 because no item had an extractable primary key", resource, consumedTotal),
			Duration: time.Since(started),
		}
	}

	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(resource string) paginationDefaults {
	switch resource {
	case "items":
		return paginationDefaults{
			cursorParam: "cursor",
			cursorType:  "cursor",
			limitParam:  "limit",
			limit:       100,
		}
	}
	return paginationDefaults{
		cursorParam: "cursor",
		cursorType:  "cursor",
		limitParam:  "limit",
		limit:       100,
	}
}

func resourceSupportsPagination(resource string) bool {
	switch resource {
	case "items":
		return true
	}
	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 ""
}

func syncResourceSinceParamFormat(resource string) string {
	switch resource {
	}
	return ""
}

func formatSyncSinceValue(value string, paramFormat string) string {
	if field, ok := strings.CutPrefix(paramFormat, "odata-conditions:"); ok {
		field = strings.TrimSpace(field)
		if field == "" {
			return value
		}
		return fmt.Sprintf("%s > [%s]", field, value)
	}
	if strings.EqualFold(paramFormat, "date") {
		if ts, err := time.Parse(time.RFC3339, value); err == nil {
			return ts.Format("2006-01-02")
		}
		if ts, err := time.Parse(time.RFC3339Nano, value); err == nil {
			return ts.Format("2006-01-02")
		}
		if _, err := time.Parse("2006-01-02", value); err == nil {
			return value
		}
	}
	return value
}

// 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"
// 3. JSend-style nested data envelopes: {"data":{"<resource>":[...]}}
// It also extracts the next cursor from common response fields.
func extractPageItems(data json.RawMessage, cursorParam string, responsePaths ...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
	}

	for _, responsePath := range responsePaths {
		pathData, ok := responsePayloadAtPath(data, responsePath)
		if !ok {
			continue
		}
		if items, ok := extractObjectArray(pathData); ok {
			nextCursor, hasMore := "", false
			if parentEnvelope, ok := responsePayloadParentAtPath(data, responsePath); ok {
				nextCursor, hasMore = extractPaginationFromEnvelope(parentEnvelope, cursorParam)
			}
			outerCursor, outerHasMore := extractPaginationFromEnvelope(envelope, cursorParam)
			if nextCursor == "" {
				nextCursor = outerCursor
			}
			hasMore = hasMore || outerHasMore
			return items, nextCursor, hasMore
		}
		var inner map[string]json.RawMessage
		if json.Unmarshal(pathData, &inner) == nil {
			if items, ok := extractItemsFromEnvelope(inner); ok {
				nextCursor, hasMore := extractPaginationFromEnvelope(inner, cursorParam)
				outerCursor, outerHasMore := extractPaginationFromEnvelope(envelope, cursorParam)
				if nextCursor == "" {
					nextCursor = outerCursor
				}
				hasMore = hasMore || outerHasMore
				return items, nextCursor, hasMore
			}
		}
	}

	if items, ok := extractItemsByKnownKeys(envelope); ok {
		nextCursor, hasMore := extractPaginationFromEnvelope(envelope, cursorParam)
		return items, nextCursor, hasMore
	}

	for _, key := range dataEnvelopeKeys {
		raw, ok := envelope[key]
		if !ok {
			continue
		}
		var inner map[string]json.RawMessage
		if json.Unmarshal(raw, &inner) != nil {
			continue
		}
		if items, ok := extractItemsFromEnvelope(inner); ok {
			nextCursor, hasMore := extractPaginationFromEnvelope(inner, cursorParam)
			outerCursor, outerHasMore := extractPaginationFromEnvelope(envelope, cursorParam)
			if nextCursor == "" {
				nextCursor = outerCursor
			}
			hasMore = hasMore || outerHasMore
			return items, nextCursor, hasMore
		}
	}

	if raw, ok := envelope["_embedded"]; ok {
		var embedded map[string]json.RawMessage
		if json.Unmarshal(raw, &embedded) == nil {
			if items, ok := extractItemsFromEnvelope(embedded); ok {
				nextCursor, hasMore := extractPaginationFromEnvelope(envelope, cursorParam)
				return items, nextCursor, hasMore
			}
		}
	}

	if items, ok := extractSingleObjectArraySibling(envelope); ok {
		nextCursor, hasMore := extractPaginationFromEnvelope(envelope, cursorParam)
		return items, nextCursor, hasMore
	}

	return nil, "", false
}

func extractItemsFromEnvelope(envelope map[string]json.RawMessage) ([]json.RawMessage, bool) {
	if items, ok := extractItemsByKnownKeys(envelope); ok {
		return items, true
	}
	return extractSingleObjectArraySibling(envelope)
}

func extractItemsByKnownKeys(envelope map[string]json.RawMessage) ([]json.RawMessage, bool) {
	for _, key := range pageItemKeys {
		if raw, ok := envelope[key]; ok {
			if items, ok := extractObjectArray(raw); ok {
				return items, true
			}
		}
	}
	return nil, false
}

func extractSingleObjectArraySibling(envelope map[string]json.RawMessage) ([]json.RawMessage, bool) {
	// 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 alongside arbitrary scalar metadata
	// (e.g., {"markets": [...], "request_id": "..."}).
	var arrayItems []json.RawMessage
	arrayCount := 0
	for key, raw := range envelope {
		if pageMetadataArrayKeys[key] {
			continue
		}
		if candidate, ok := extractObjectArray(raw); ok {
			arrayItems = candidate
			arrayCount++
			continue
		}
		var rawArray []json.RawMessage
		if json.Unmarshal(raw, &rawArray) == nil && !isJSONNull(raw) {
			continue
		}
	}
	if arrayCount == 1 {
		return arrayItems, true
	}
	return nil, false
}

func extractObjectArray(raw json.RawMessage) ([]json.RawMessage, bool) {
	var items []json.RawMessage
	if err := json.Unmarshal(raw, &items); err != nil || len(items) == 0 {
		return nil, false
	}
	var obj map[string]json.RawMessage
	if err := json.Unmarshal(items[0], &obj); err != nil {
		return nil, false
	}
	return items, true
}

// 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, responsePaths ...string) 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
	}

	if isEmptyPageEnvelope(envelope) {
		return true
	}

	for _, responsePath := range responsePaths {
		pathData, ok := responsePayloadAtPath(data, responsePath)
		if !ok {
			continue
		}
		if isJSONNull(pathData) {
			if envelopeReportsFailure(envelope) {
				return true
			}
			continue
		}
		var direct []json.RawMessage
		if json.Unmarshal(pathData, &direct) == nil && !isJSONNull(pathData) {
			return len(direct) == 0
		}
		var inner map[string]json.RawMessage
		if json.Unmarshal(pathData, &inner) == nil && isEmptyPageEnvelope(inner) {
			return true
		}
	}

	for _, key := range dataEnvelopeKeys {
		raw, ok := envelope[key]
		if !ok {
			continue
		}
		if isJSONNull(raw) {
			if envelopeReportsFailure(envelope) {
				return true
			}
			continue
		}
		var inner map[string]json.RawMessage
		if json.Unmarshal(raw, &inner) == nil && isEmptyPageEnvelope(inner) {
			return true
		}
	}

	return false
}

func envelopeReportsFailure(envelope map[string]json.RawMessage) bool {
	for _, key := range []string{"success", "Success"} {
		if raw, ok := envelope[key]; ok {
			var success bool
			if json.Unmarshal(raw, &success) == nil {
				return !success
			}
		}
	}
	for _, key := range []string{"status", "Status"} {
		if raw, ok := envelope[key]; ok {
			var status string
			if json.Unmarshal(raw, &status) == nil {
				switch strings.ToLower(status) {
				case "error", "fail", "failed":
					return true
				}
				return false
			}
		}
	}
	return false
}

func isEmptyPageEnvelope(envelope map[string]json.RawMessage) bool {
	for _, key := range pageItemKeys {
		if raw, ok := envelope[key]; ok {
			var items []json.RawMessage
			if err := json.Unmarshal(raw, &items); err == nil && !isJSONNull(raw) {
				if len(items) > 0 {
					var obj map[string]json.RawMessage
					if err := json.Unmarshal(items[0], &obj); err != nil {
						continue
					}
				}
				return len(items) == 0
			}
		}
	}
	if hasExactlyOneNullArrayWithZeroCount(envelope) {
		return true
	}
	return hasExactlyOneEmptyArray(envelope)
}

func hasExactlyOneNullArrayWithZeroCount(envelope map[string]json.RawMessage) bool {
	nullArrayCount := 0
	hasZeroCount := false
	for key, raw := range envelope {
		if isZeroCountField(key, raw) {
			hasZeroCount = true
			continue
		}
		if isJSONNull(raw) && isNullPageItemCandidateKey(key) {
			nullArrayCount++
			continue
		}
		if pageEnvelopeMetadataKeys[key] {
			continue
		}
		return false
	}
	return nullArrayCount == 1 && hasZeroCount
}

func isZeroCountField(key string, raw json.RawMessage) bool {
	switch key {
	case "total", "Total", "count", "Count", "total_count", "totalCount", "TotalCount":
	default:
		return false
	}
	// A JSON null count is not a numeric zero-count signal: json.Unmarshal of
	// "null" into a float64 is a no-op (leaves n at 0, returns nil), so without
	// this guard a null count would falsely qualify as zero — masking a
	// possibly-malformed {"items":null,"total":null} as an empty page.
	if isJSONNull(raw) {
		return false
	}
	var n float64
	return json.Unmarshal(raw, &n) == nil && n == 0
}

func isNullPageItemCandidateKey(key string) bool {
	for _, itemKey := range pageItemKeys {
		if key == itemKey {
			return true
		}
	}
	return !pageEnvelopeMetadataKeys[key]
}

func hasExactlyOneEmptyArray(envelope map[string]json.RawMessage) bool {
	arrayCount := 0
	for _, raw := range envelope {
		var candidate []json.RawMessage
		if err := json.Unmarshal(raw, &candidate); err == nil && !isJSONNull(raw) {
			// Skip candidate arrays that contain primitives (not objects).
			if len(candidate) > 0 {
				var obj map[string]json.RawMessage
				if err := json.Unmarshal(candidate[0], &obj); err != nil {
					continue
				}
			}
			if len(candidate) == 0 {
				arrayCount++
			}
		}
	}
	return arrayCount == 1
}

func isJSONNull(raw json.RawMessage) bool {
	return strings.TrimSpace(string(raw)) == "null"
}

func isJSONResponse(data json.RawMessage) bool {
	var probe any
	return json.Unmarshal(data, &probe) == nil
}

// 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", "next_token", "nextToken", "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":"..."}}; Pipedrive uses
	// "additional_data"; 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", "additional_data", "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
			}
		}
	}

	if nextCursor == "" {
		nextCursor = nextCursorFromTopLevelURL(envelope, cursorParam)
	}

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

func pageAllowsPageIntFallback(data json.RawMessage) bool {
	return pageMayHaveMore(data)
}

func pageMayHaveMore(data json.RawMessage, responsePaths ...string) bool {
	hasMore, parsed := pageExplicitHasMore(data, responsePaths...)
	return !parsed || hasMore
}

func pageExplicitHasMore(data json.RawMessage, responsePaths ...string) (bool, bool) {
	var envelope map[string]json.RawMessage
	if err := json.Unmarshal(data, &envelope); err != nil {
		return false, false
	}
	if hasMore, parsed := envelopeExplicitHasMore(envelope); parsed {
		return hasMore, true
	}
	for _, responsePath := range responsePaths {
		pathData, ok := responsePayloadAtPath(data, responsePath)
		if !ok {
			continue
		}
		var inner map[string]json.RawMessage
		if json.Unmarshal(pathData, &inner) == nil {
			if hasMore, parsed := envelopeExplicitHasMore(inner); parsed {
				return hasMore, true
			}
		}
	}
	for _, key := range dataEnvelopeKeys {
		raw, ok := envelope[key]
		if !ok {
			continue
		}
		var inner map[string]json.RawMessage
		if json.Unmarshal(raw, &inner) == nil {
			if hasMore, parsed := envelopeExplicitHasMore(inner); parsed {
				return hasMore, true
			}
		}
	}
	return false, false
}

func envelopeExplicitHasMore(envelope map[string]json.RawMessage) (bool, bool) {
	for _, key := range []string{"has_more", "hasMore", "has_next", "hasNext", "next_page"} {
		raw, ok := envelope[key]
		if !ok {
			continue
		}
		var hasMore bool
		if json.Unmarshal(raw, &hasMore) == nil {
			return hasMore, true
		}
	}
	return false, false
}

// nextCursorFromLinks extracts pagination cursors from JSON:API
// {"links":{"next":"https://example.com/items?page[cursor]=..."}} and HAL
// {"_links":{"next":{"href":"https://example.com/items?cursor=..."}}}.
func nextCursorFromLinks(envelope map[string]json.RawMessage, cursorParam string) string {
	for _, key := range []string{"links", "_links"} {
		rawLinks, ok := envelope[key]
		if !ok {
			continue
		}
		var links map[string]json.RawMessage
		if json.Unmarshal(rawLinks, &links) != nil {
			continue
		}
		rawNext, ok := links["next"]
		if !ok {
			continue
		}
		if nextURL := paginationLinkURL(rawNext); nextURL != "" {
			if cursor := cursorFromNextURL(nextURL, cursorParam); cursor != "" {
				return cursor
			}
		}
	}
	return ""
}

func paginationLinkURL(raw json.RawMessage) string {
	var nextURL string
	if json.Unmarshal(raw, &nextURL) == nil {
		return nextURL
	}
	var link map[string]json.RawMessage
	if json.Unmarshal(raw, &link) != nil {
		return ""
	}
	rawHref, ok := link["href"]
	if !ok {
		return ""
	}
	if json.Unmarshal(rawHref, &nextURL) != nil {
		return ""
	}
	return nextURL
}

// nextCursorFromTopLevelURL extracts a cursor from top-level absolute or relative next URLs.
func nextCursorFromTopLevelURL(envelope map[string]json.RawMessage, cursorParam string) string {
	for _, key := range []string{"next", "next_url"} {
		rawNext, ok := envelope[key]
		if !ok {
			continue
		}
		var nextURL string
		if json.Unmarshal(rawNext, &nextURL) != nil || nextURL == "" {
			continue
		}
		if isFollowableNextURL(nextURL) {
			return cursorFromNextURL(nextURL, cursorParam)
		}
	}
	return ""
}

// isFollowableNextURL reports whether a top-level "next" string is a URL we can
// pull a cursor query param from: an absolute http(s) URL, a root-relative path,
// or any value carrying a query string. Bare opaque cursor tokens (no query)
// are rejected so they aren't mis-parsed.
func isFollowableNextURL(nextURL string) bool {
	lower := strings.ToLower(nextURL)
	return strings.HasPrefix(lower, "http") ||
		strings.HasPrefix(nextURL, "/") ||
		strings.Contains(nextURL, "?")
}

func cursorFromNextURL(nextURL string, cursorParam string) string {
	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 ""
}

func emitSyncMissingPaginationCursorWarning(syncEvents io.Writer, humanFriendly bool, resource, parent string) {
	if humanFriendly {
		if parent != "" {
			fmt.Fprintf(os.Stderr, "\nwarning: %s returned a full page for parent %s without a next cursor; data may be truncated.\n", resource, parent)
			return
		}
		fmt.Fprintf(os.Stderr, "\nwarning: %s returned a full page without a next cursor; data may be truncated.\n", resource)
		return
	}
	if parent != "" {
		fmt.Fprintf(syncEvents, `{"event":"sync_warning","resource":"%s","parent":"%s","reason":"pagination_cursor_missing","message":"API returned a full page without a usable next cursor; data may be truncated."}`+"\n", resource, parent)
		return
	}
	fmt.Fprintf(syncEvents, `{"event":"sync_warning","resource":"%s","reason":"pagination_cursor_missing","message":"API returned a full page without a usable next cursor; data may be truncated."}`+"\n", resource)
}

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
		if obj, err := store.DecodeJSONObject(item); 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 {
	obj, err := store.DecodeJSONObject(data)
	if 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 {
	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{
		"items",
	}
}

// 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 := []string{
		"items",
	}
	return names
}

func describeFailedResources(count int, resources []string) string {
	return describeResourceFailure(count, "resource(s)", resources)
}

func describeCriticalFailedResources(count int, resources []string) string {
	return describeResourceFailure(count, "critical resource(s)", resources)
}

func describeResourceFailure(count int, label string, resources []string) string {
	if len(resources) == 0 {
		return fmt.Sprintf("%d %s failed to sync", count, label)
	}
	names := append([]string(nil), resources...)
	sort.Strings(names)
	return fmt.Sprintf("%d %s failed to sync: %s", count, label, strings.Join(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{
		"items": "/items",
	}
	if p, ok := paths[resource]; ok {
		return p, nil
	}
	return "", fmt.Errorf("unknown sync resource %q", resource)
}

var syncResourceTiers = map[string]string{
	"items": "free",
}

func syncClientForResource(c *client.Client, resource string) *client.Client {
	if tier, ok := syncResourceTiers[resource]; ok {
		return c.WithTier(tier)
	}
	return c
}

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

// partitionOutcome tracks whether a sync loop (flat tenant-scoped OR dependent
// per-parent) enumerated its partition completely. complete is set ONLY at
// proven natural ends; any abnormal break records a reason and leaves
// complete=false so the gated reconcile SKIPS. scopeVal carries the dependent
// loop's parent scope value (unused by the flat path).
type partitionOutcome struct {
	complete bool
	reason   string
	scopeVal string
}

// flatReconcileModes maps a flat resource to its reconcile mode classification
// (from SyncableResource.ReconcileMode, set by the profiler when a flat resource
// carries a tenant scope column AND an extractable IDField AND no discriminator).
// Only "flat" resources are emitted; resourceReconcileMode returns "" for any
// resource absent here, which is all the flat reconcile gate checks for.
var flatReconcileModes = map[string]string{}

// resourceReconcileMode returns the flat reconcile classification for a resource,
// or "" when it is not a flat-reconcilable resource.
func resourceReconcileMode(resource string) string {
	return flatReconcileModes[resource]
}

// flatReconcileDefT carries the per-resource metadata the flat reconcile call
// site needs. BodyField is the JSON body key (== TenantScopeColumn; the column
// name and the body key coincide) used to scope ReconcilePartition's
// json_extract($.<BodyField>) partition filter to the active tenant.
type flatReconcileDefT struct {
	BodyField string
}

// flatReconcileDefs maps each flat-reconcilable resource to its tenant body
// field. Sourced from SyncableResource.TenantScopeColumn for ReconcileMode=="flat".
var flatReconcileDefs = map[string]flatReconcileDefT{}

// flatReconcileDef returns the flat reconcile metadata for a resource (zero
// value when absent; the call site only reaches it for flatReconcilable resources).
func flatReconcileDef(resource string) flatReconcileDefT {
	return flatReconcileDefs[resource]
}

// resolveTenantID returns the active tenant's stable ID for tenant-scoped
// enumeration and reconcile, or "" = unknown. The generated default cannot
// resolve (a printed CLI has no tenant registry). A novel override reassigns
// this with a real resolver in RunE (after flag parsing). No-arg by design: it
// must not reference any novel (hand-patched) type, so the override owns all
// config/slug access internally. Consumers branch on "" with OPPOSITE fallbacks:
// fan-out falls back to UNSCOPED enumeration; flat reconcile SKIPS (never
// deletes).
var resolveTenantID = func() string { return "" }

// parentTenantScopeColumns maps a dependent-parent table to its tenant
// discriminator column (from x-pp-tenant-scope-column), used to scope dependent
// fan-out. Empty when no parent is annotated.
var parentTenantScopeColumns = map[string]string{}

// 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", "api_id", "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", "features",
	"Data", "Results", "Items", "Records", "Nodes", "Entries", "Features",
}

var dataEnvelopeKeys = []string{"data", "Data", "result", "Result"}

func responsePathForResource(resource, path string) []string {
	switch resource + "\x00" + path {
	}
	return nil
}

var pageMetadataArrayKeys = map[string]bool{
	"errors": true, "Errors": true,
	"warnings": true, "Warnings": true,
}

var pageEnvelopeMetadataKeys = map[string]bool{
	// list wrappers themselves
	"data": true, "results": true, "items": true,
	"Data": true, "Results": true, "Items": true,
	// pagination cursors / tokens
	"next_cursor": true, "nextCursor": true, "NextCursor": true,
	"next_token": true, "nextToken": true, "NextToken": true,
	"next_page_token": true, "nextPageToken": true, "NextPageToken": true,
	"page_token": true, "pageToken": true, "PageToken": true,
	"end_cursor": true, "endCursor": true, "EndCursor": true,
	"start_cursor": true, "startCursor": true, "StartCursor": true,
	"cursor": true, "Cursor": true, "after": true, "After": true, "before": true, "Before": true,
	// has-more flags and page numbers
	"has_more": true, "hasMore": true, "HasMore": true,
	"has_next": true, "hasNext": true, "HasNext": true,
	"next_page": true, "nextPage": true, "NextPage": true,
	"previous_page": true, "previousPage": true, "PreviousPage": true,
	"page": true, "Page": true, "page_size": true, "pageSize": true, "PageSize": true,
	"per_page": true, "perPage": true, "PerPage": true,
	// counts / totals
	"total": true, "Total": true, "count": true, "Count": true, "size": true, "Size": true,
	"total_count": true, "totalCount": true, "TotalCount": true,
	// JSend / common status envelopes
	"success": true, "status": true, "message": true, "error": true, "errors": true,
	"warnings": true, "Warnings": true, "ok": true, "Ok": true,
	// wrapper objects
	"links": true, "meta": true, "pagination": true,
	"response_metadata": true, "paging": true,
	// links shape
	"next": true, "prev": true, "previous": true, "first": true, "last": true,
}

// 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 := store.ResourceIDString(v)
			if s != "" && s != "<nil>" {
				return s
			}
		}
	}
	for _, key := range genericIDFieldFallbacks {
		if v := store.LookupFieldValue(obj, key); v != nil {
			s := store.ResourceIDString(v)
			if s != "" && s != "<nil>" {
				return s
			}
		}
	}
	if s := suffixIDFieldFallback(resource, obj); s != "" {
		return s
	}
	return ""
}

// suffixIDFieldFallback resolves an id-less resource that keys on its own
// "<name>_code" / "<name>_id" / "<name>_key" / "<name>_slug" field (e.g. the
// "currencies" resource keying on "currency_code" — see #2327). It is scoped to
// the resource's OWN name so a foreign key like account_id/parent_id is never
// promoted to the primary key, and it uses direct map lookups in a fixed suffix
// order so the chosen id is deterministic.
func suffixIDFieldFallback(resourceType string, obj map[string]any) string {
	for _, base := range resourceIDBaseNames(resourceType) {
		for _, suffix := range []string{"_id", "_code", "_key", "_slug"} {
			if v, ok := obj[base+suffix]; ok {
				if s := scalarIDString(v); s != "" && s != "<nil>" {
					return s
				}
			}
		}
		camelBase := lowerCamelResourceIDBase(base)
		for _, suffix := range []string{"Id", "Code", "Key", "Slug"} {
			if v, ok := obj[camelBase+suffix]; ok {
				if s := scalarIDString(v); s != "" && s != "<nil>" {
					return s
				}
			}
		}
	}
	return ""
}

// resourceIDBaseNames returns lowercase candidate singular/plural stems of a
// resource name to build "<base>_id"-style key probes from (e.g. "currencies"
// -> ["currencies","currency"]). OpenAPI-/path-derived names can carry a
// leading verb token ("get-currencies"), so the same probes are also attempted
// on the de-verbed stem. Minimal English depluralization; the raw name is
// always included so already-singular names work too.
func resourceIDBaseNames(resourceType string) []string {
	r := strings.ToLower(strings.TrimSpace(resourceType))
	if r == "" {
		return nil
	}
	stems := []string{r}
	if d := stripLeadingResourceVerb(r); d != "" && d != r {
		stems = append(stems, d)
	}
	var bases []string
	seen := map[string]bool{}
	add := func(s string) {
		if s != "" && !seen[s] {
			seen[s] = true
			bases = append(bases, s)
		}
	}
	for _, stem := range stems {
		add(stem)
		add(depluralizeResourceStem(stem))
	}
	return bases
}

func stripLeadingResourceVerb(r string) string {
	for _, verb := range []string{"get", "list", "fetch", "find", "retrieve", "read", "show", "all"} {
		for _, sep := range []string{"-", "_"} {
			prefix := verb + sep
			if strings.HasPrefix(r, prefix) && len(r) > len(prefix) {
				return r[len(prefix):]
			}
		}
	}
	return ""
}

func depluralizeResourceStem(r string) string {
	switch {
	case strings.HasSuffix(r, "ies") && len(r) > 3:
		return strings.TrimSuffix(r, "ies") + "y" // currencies -> currency
	// Plurals formed by adding "es" to a base ending in s/x/z/ch/sh. The
	// double-s "sses" guard (not bare "ses") keeps soft-e plurals — where the
	// singular already ends in a silent "e" (cases, databases, licenses,
	// purchases) — out of this branch; they fall through to the "-s" case below
	// (cases -> case, not cas). Trade-off: a genuine "-es" plural of an s-ending
	// singular (buses, statuses) depluralizes imperfectly, but those are rare as
	// resource names and this stem only feeds best-effort id-field probing.
	case strings.HasSuffix(r, "sses") || strings.HasSuffix(r, "xes") ||
		strings.HasSuffix(r, "zes") || strings.HasSuffix(r, "ches") ||
		strings.HasSuffix(r, "shes"):
		return strings.TrimSuffix(r, "es") // classes -> class, boxes -> box, dishes -> dish
	case strings.HasSuffix(r, "s") && !strings.HasSuffix(r, "ss") && len(r) > 1:
		return strings.TrimSuffix(r, "s") // languages -> language, cases -> case
	}
	return r
}

func lowerCamelResourceIDBase(base string) string {
	parts := strings.FieldsFunc(base, func(r rune) bool {
		return r == '_' || r == '-'
	})
	if len(parts) == 0 {
		return base
	}
	for i := range parts {
		if i == 0 {
			parts[i] = strings.ToLower(parts[i])
			continue
		}
		parts[i] = strings.ToUpper(parts[i][:1]) + strings.ToLower(parts[i][1:])
	}
	return strings.Join(parts, "")
}

func scalarIDString(value any) string {
	switch value.(type) {
	case string, bool, int, int8, int16, int32, int64,
		uint, uint8, uint16, uint32, uint64,
		float32, float64, json.Number, []byte:
		return store.ResourceIDString(value)
	default:
		return ""
	}
}
