// Server-side recycle-bin LIST aggregator.
//
// Replaces the prior client-side per-SObject fan-out against /queryAll
// (see pingcap-inc/NexusCRM#1681). Does what the frontend used to do, but
// on the server, in two phases, so that delete-event ownership and record
// visibility each come from their own authoritative source:
//
//  1. Delete-event authority — `_System_RecycleBin` tombstones. Each soft
//     delete writes one tombstone row carrying the immutable delete facts:
//     who deleted the record (the stable DeletedById, plus a DeletedBy
//     display-name snapshot), when (DeletedDate), the record's object + id +
//     name snapshot. scope=mine selects the caller's own tombstones by
//     DeletedById; scope=all selects every tombstone (admin only). Keying
//     ownership on the immutable id — not the mutable display name — is the
//     pingcap-inc/NexusCRM#1750 / #1692 fix. Legacy tombstones written before
//     DeletedById existed have no stable deleter id and are excluded from
//     scope=mine (visible via scope=all only).
//
//  2. Record visibility — `QueryService.QueryWithSpec`. The tombstone is an
//     index, NOT a secured read path: we never trust it to decide what the
//     caller may see. For each object we re-read the deleted rows by id
//     through QueryService (`Id IN (...) AND IsDeleted=true`, ScanAll). For
//     scope=mine this is an ordinary user-context read. For scope=all, after
//     the ViewAllData/ModifyAllData gate, the per-object read uses the narrow
//     scoped-internal-read primitive that QueryService already honors for
//     object-read/RLS bypass while still blocking hard-blocked objects. Only
//     rows that survive the relevant QueryService path are returned. This is
//     why #1687's "read straight from `_System_RecycleBin`" was wrong and
//     this shape is not.
//
// The two are joined per row: delete-event facts (DeletedBy / DeletedDate)
// from the tombstone, record facts (name / CreatedDate / LastModifiedDate)
// from the FLS-gated live read.
//
// Design rationale and decisions: pingcap-inc/NexusCRM#1722, #1738.
package records

import (
	"context"
	"errors"
	"fmt"
	"log"
	"net/http"
	"sort"
	"strings"
	"sync"
	"time"

	"github.com/nexuscrm/kernel/pkg/apierr"
	"github.com/nexuscrm/kernel/pkg/ports"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
	"github.com/nexuscrm/shared/pkg/queryspec"
)

// RecycleBinScope is the public scope discriminator for the LIST API.
type RecycleBinScope string

const (
	// RecycleBinScopeMine returns the caller's own deleted records: the
	// records whose `_System_RecycleBin` tombstone stamps the caller's stable
	// DeletedById, intersected with the records the caller can still read
	// through the standard query path (object-read + RLS + FLS).
	RecycleBinScopeMine RecycleBinScope = "mine"
	// RecycleBinScopeAll returns every deleted record, gated on
	// CheckSystemPermission(ViewAllData) OR CheckSystemPermission(ModifyAllData)
	// — system-permission only, no custom-permission shadow. Visibility is
	// still re-checked per record through QueryService.
	RecycleBinScopeAll RecycleBinScope = "all"
)

const (
	// Aggregate row cap. This is a post-visibility result cap: tombstone
	// candidates are paged, each page is re-read through QueryService, and only
	// visible rows count against the published 200-row LIST limit.
	recycleBinAggregateLimit = 200

	recycleBinCandidatePageSize  = recycleBinAggregateLimit
	recycleBinCandidateScanLimit = recycleBinAggregateLimit * 10

	// Concurrent per-SObject query budget. Hard-coded for v1: 8 in-flight
	// is comfortably under the per-Connected-App rate limit and well below
	// the TiDB Serverless connection pool. If a tenant reports timeouts or
	// contention, lift this to a NEXUS_RECYCLEBIN_FANOUT_PARALLELISM env
	// var (pattern matches NEXUS_API_HOURLY_LIMIT in pkg/serverruntime/routes.go).
	recycleBinFanoutParallelism = 8

	// Recycle-bin LIST is not a standard action. This deliberately reuses the
	// existing scoped-internal-read context shape because QueryService already
	// honors it for narrow object-read/RLS bypasses while preserving blocked
	// object checks.
	recycleBinScopedReadActionID = "recycle-bin-scope-all"
)

// RecycleBinItem is the normalized per-deleted-record projection returned
// to the handler. Field names match the existing frontend RecycleBinItem
// TypeScript type so the FE wire shape stays stable across the migration.
type RecycleBinItem struct {
	ID               string
	RecordID         string
	RecordName       string
	ObjectAPIName    string
	DeletedBy        string
	DeletedDate      string
	CreatedDate      string
	LastModifiedDate string

	// tombstoneID is the _System_RecycleBin.Id, carried for the final
	// (DeletedDate DESC, Id DESC) ordering that makes the aggregate cap
	// deterministic for same-second deletes. Not serialized to the wire.
	tombstoneID string
}

// RecycleBinFailure carries the information needed for the handler to
// surface a partial-failure signal to the client.
type RecycleBinFailure struct {
	ObjectAPIName string
	ErrorCode     string
	Message       string
}

// RecycleBinListResult is the service-layer return shape. The handler is
// responsible for mapping to the v63 wire envelope.
type RecycleBinListResult struct {
	Items         []RecycleBinItem
	FailedObjects []RecycleBinFailure
}

// SetQueryService wires the canonical user-authorized query path used by the
// recycle-bin fan-out and data-derived create-default previews. It follows the
// same lazy-setter pattern as the
// other optional dependencies on PersistenceService (SetRawDB,
// SetActiveFormulaProjector) so the constructor signature stays stable.
//
// Callers that require either capability fail closed when it is not wired;
// neither path may fall back to a raw repository read.
func (ps *PersistenceService) SetQueryService(svc ports.QueryService) {
	ps.querySvc = svc
}

// GetRecycleBinItems returns the recycle-bin LIST as the calling user can
// see it, filtered by scope. See package doc and pingcap-inc/NexusCRM#1722,
// #1738 for design rationale.
func (ps *PersistenceService) GetRecycleBinItems(ctx context.Context, user *models.UserSession, scope RecycleBinScope) (*RecycleBinListResult, error) {
	if user == nil {
		return nil, apierr.Invalid("user session required")
	}
	if ps.querySvc == nil {
		return nil, apierr.Internal(fmt.Errorf("recycle-bin: query service not wired (call SetQueryService at startup)"))
	}

	switch scope {
	case RecycleBinScopeMine, RecycleBinScopeAll:
	default:
		return nil, apierr.Invalid(fmt.Sprintf("unsupported scope %q (expected %q or %q)", scope, RecycleBinScopeMine, RecycleBinScopeAll))
	}

	if scope == RecycleBinScopeAll && !ps.callerCanViewAllRecycleBin(ctx, user) {
		return nil, apierr.Forbidden("ViewAllData or ModifyAllData required to view the org recycle bin")
	}

	// Wiring check deferred until after request validation so a malformed
	// scope / forbidden scope=all still returns 400/403 rather than 500.
	if ps.repo == nil {
		return nil, apierr.Internal(fmt.Errorf("recycle-bin: record repository not wired"))
	}

	items := make([]RecycleBinItem, 0, recycleBinAggregateLimit)
	failureByObject := make(map[string]RecycleBinFailure)
	failureOrder := make([]string, 0)
	cursor := recycleBinPageCursor{}
	candidatesScanned := 0
	lastPageFull := false

	for len(items) < recycleBinAggregateLimit && candidatesScanned < recycleBinCandidateScanLimit {
		pageLimit := min(recycleBinCandidatePageSize, recycleBinCandidateScanLimit-candidatesScanned)

		// Phase 1: delete-event authority.
		tombstones, nextCursor, rawCount, exhausted, err := ps.loadDeleteEventAuthorityPage(ctx, user, scope, cursor, pageLimit)
		if err != nil {
			return nil, err
		}
		if rawCount == 0 {
			break
		}
		candidatesScanned += rawCount
		lastPageFull = rawCount == pageLimit && !exhausted
		cursor = nextCursor

		// Phase 2: record visibility via QueryService.
		groups, err := ps.planRecycleBinGroups(ctx, user, scope, tombstones)
		if err != nil {
			return nil, err
		}
		pageItems, pageFailures, err := ps.fanOutRecycleBinVisibility(ctx, user, scope, groups)
		if err != nil {
			return nil, err
		}
		items = append(items, pageItems...)
		for _, failure := range pageFailures {
			if _, exists := failureByObject[failure.ObjectAPIName]; exists {
				continue
			}
			failureByObject[failure.ObjectAPIName] = failure
			failureOrder = append(failureOrder, failure.ObjectAPIName)
		}

		if exhausted {
			break
		}
	}

	sortItemsByDeletedDateDesc(items)
	if len(items) > recycleBinAggregateLimit {
		items = items[:recycleBinAggregateLimit]
	}
	if candidatesScanned >= recycleBinCandidateScanLimit && lastPageFull && len(items) < recycleBinAggregateLimit {
		log.Printf("recycle-bin candidate scan truncated: scope=%s scanned=%d visible=%d scanLimit=%d", scope, candidatesScanned, len(items), recycleBinCandidateScanLimit)
	}

	failures := make([]RecycleBinFailure, 0, len(failureOrder))
	for _, objectAPIName := range failureOrder {
		failures = append(failures, failureByObject[objectAPIName])
	}
	return &RecycleBinListResult{Items: items, FailedObjects: failures}, nil
}

// callerCanViewAllRecycleBin gates scope=all on ViewAllData OR
// ModifyAllData. CheckSystemPermission is used in preference to the
// general-purpose HasPermission because HasPermission also matches
// custom permissions sharing the same apiName: a tenant admin who
// creates a custom permission literally named "ViewAllData" would
// otherwise unlock the org recycle bin. record_access.go and
// rls_filter.go gate the same pair of permissions through
// CheckSystemPermission for the same reason.
func (ps *PersistenceService) callerCanViewAllRecycleBin(ctx context.Context, user *models.UserSession) bool {
	if ps.permissions == nil {
		return false
	}
	if ps.permissions.CheckSystemPermission(ctx, user, string(constants.PermViewAllData)) {
		return true
	}
	return ps.permissions.CheckSystemPermission(ctx, user, string(constants.PermModifyAllData))
}

// recycleBinTombstone is one normalized `_System_RecycleBin` row: the
// immutable delete-event facts plus the keys needed to re-read the live
// record for the visibility check.
type recycleBinTombstone struct {
	id            string // _System_RecycleBin.Id — the keyset tie-break for stable ordering
	recordID      string
	objectAPIName string
	recordName    string // snapshot at delete time; fallback display name only
	deletedBy     string
	deletedDate   string
}

type recycleBinPageCursor struct {
	deletedDate string
	id          string
}

// loadDeleteEventAuthorityPage reads one page of the caller's relevant
// tombstones from `_System_RecycleBin`. scope=mine matches DeletedById against
// the caller's stable id — the same key the soft-delete path writes
// (persistence_delete.go). scope=all loads every tombstone (already gated on
// ViewAll/ModifyAll).
func (ps *PersistenceService) loadDeleteEventAuthorityPage(ctx context.Context, user *models.UserSession, scope RecycleBinScope, cursor recycleBinPageCursor, limit int) ([]recycleBinTombstone, recycleBinPageCursor, int, bool, error) {
	if limit <= 0 {
		return nil, cursor, 0, true, nil
	}
	var (
		rows []models.SObject
		err  error
	)
	switch scope {
	case RecycleBinScopeMine:
		// scope=mine matches the caller's own tombstones by the stable
		// DeletedById (written from currentUser.ID at delete time). Keying on
		// the immutable id — never the mutable display Name — is the fix for the
		// same-name collision and rename orphaning of pingcap-inc/NexusCRM#1750
		// / #1692. Legacy tombstones written before DeletedById existed
		// (DeletedById IS NULL) have no authoritative delete-time identity and
		// are excluded from scope=mine (visible via scope=all); see
		// FindRecycleBinItemsByUserPage. Visibility is still re-checked per
		// record through QueryService below.
		//
		// A caller with no stable id has no key to match, so their bin is empty
		// rather than matching blank-id tombstones.
		deleterID := strings.TrimSpace(user.ID)
		if deleterID == "" {
			return nil, cursor, 0, true, nil
		}
		rows, err = ps.repo.FindRecycleBinItemsByUserPage(ctx, deleterID, cursor.deletedDate, cursor.id, limit)
	case RecycleBinScopeAll:
		rows, err = ps.repo.FindAllRecycleBinItemsPage(ctx, cursor.deletedDate, cursor.id, limit)
	}
	if err != nil {
		// Wrap as internal: the handler collapses 5xx bodies to a generic
		// message, so the raw repo/driver text is logged, not returned.
		return nil, cursor, 0, true, apierr.Internal(err)
	}

	rawCount := len(rows)
	next := cursor
	if rawCount > 0 {
		lastRaw := rows[rawCount-1]
		next = recycleBinPageCursor{
			// DATETIME literal for the keyset predicate, NOT stringValue's
			// time.Time.String() — see dbDateTime. The Id is already a string.
			deletedDate: dbDateTime(lastRaw[constants.FieldSysRecycleBin_DeletedDate]),
			id:          stringValue(lastRaw[constants.FieldSysRecycleBin_ID]),
		}
	}

	out := make([]recycleBinTombstone, 0, len(rows))
	for _, r := range rows {
		recordID := stringValue(r[constants.FieldRecordID])
		objectAPIName := stringValue(r[constants.FieldObjectAPIName])
		if recordID == "" || objectAPIName == "" {
			continue
		}
		out = append(out, recycleBinTombstone{
			id:            stringValue(r[constants.FieldSysRecycleBin_ID]),
			recordID:      recordID,
			objectAPIName: objectAPIName,
			recordName:    stringValue(r[constants.FieldRecordName]),
			deletedBy:     stringValue(r[constants.FieldDeletedBy]),
			deletedDate:   wireTimestamp(r[constants.FieldDeletedDate]),
		})
	}
	return out, next, rawCount, rawCount < limit, nil
}

// recycleBinObjectGroup is the per-object visibility-query plan: the record
// ids to re-read (as []interface{} for the IN predicate), the name field to
// project, and the tombstone facts to merge back in keyed by record id.
type recycleBinObjectGroup struct {
	apiName           string
	nameField         string
	nameFieldReadable bool
	recordIDs         []interface{}
	tombstones        map[string]recycleBinTombstone
}

// planRecycleBinGroups groups tombstones by object, resolves each object's
// name field from live metadata, and (for scope=mine only) applies the
// object-read pre-filter.
//
// scope=mine: the pre-filter (CheckObjectPermissionWithUser Read) drops an
// object the caller cannot read from the plan entirely — it never reaches
// the fan-out, so it can never surface as a failedObjects entry. That keeps
// the partial-failure payload from disclosing the API names of objects a
// regular user is not authorized to know exist (#1738 finding #2).
//
// scope=all: NO pre-filter. This scope is already gated on ViewAllData /
// ModifyAllData, so the caller is privileged and object-name disclosure is
// not a concern. More importantly, CheckObjectPermissionWithUser does not
// treat ViewAllData/ModifyAllData as a general object-read bypass, so
// pre-filtering here would
// SILENTLY drop objects a ViewAllData-only caller is entitled to see under
// the advertised "whole org bin" contract. The later per-object
// QueryService read carries a narrow scoped internal read for object/RLS
// bypass while blocked-object checks still run.
func (ps *PersistenceService) planRecycleBinGroups(ctx context.Context, user *models.UserSession, scope RecycleBinScope, tombstones []recycleBinTombstone) ([]recycleBinObjectGroup, error) {
	type accumulator struct {
		ids  []interface{}
		tomb map[string]recycleBinTombstone
	}
	byObject := map[string]*accumulator{}
	order := make([]string, 0)
	for _, t := range tombstones {
		acc := byObject[t.objectAPIName]
		if acc == nil {
			acc = &accumulator{tomb: map[string]recycleBinTombstone{}}
			byObject[t.objectAPIName] = acc
			order = append(order, t.objectAPIName)
		}
		if _, seen := acc.tomb[t.recordID]; seen {
			continue
		}
		acc.ids = append(acc.ids, t.recordID)
		acc.tomb[t.recordID] = t
	}

	groups := make([]recycleBinObjectGroup, 0, len(order))
	for _, objectAPIName := range order {
		if ps.metadata == nil {
			continue
		}
		schema := ps.metadata.GetSchema(ctx, objectAPIName)
		if schema == nil {
			// Object no longer exists (dropped after delete). Nothing to
			// show and nothing to disclose.
			continue
		}
		if !isRecycleBinManagedObject(schema) {
			continue
		}
		if scope == RecycleBinScopeMine && ps.permissions != nil &&
			!ps.permissions.CheckObjectPermissionWithUser(ctx, objectAPIName, constants.PermRead, user) {
			continue
		}
		acc := byObject[objectAPIName]
		nameField := nameFieldFor(schema)
		nameFieldReadable, err := ps.recycleBinNameFieldReadable(ctx, user, objectAPIName, nameField)
		if err != nil {
			return nil, err
		}
		groups = append(groups, recycleBinObjectGroup{
			apiName:           objectAPIName,
			nameField:         nameField,
			nameFieldReadable: nameFieldReadable,
			recordIDs:         acc.ids,
			tombstones:        acc.tomb,
		})
	}
	return groups, nil
}

func (ps *PersistenceService) recycleBinNameFieldReadable(ctx context.Context, user *models.UserSession, objectAPIName string, nameField string) (bool, error) {
	if nameField == "" || strings.EqualFold(nameField, constants.FieldID) {
		return true, nil
	}
	if ps.permissions == nil {
		return true, nil
	}
	return ps.permissions.CheckFieldVisibilityWithUser(ctx, objectAPIName, nameField, user)
}

// nameFieldFor returns the API name of the schema's name field, or an
// empty string when none is declared. Used to project a human-readable
// label onto each recycle-bin row.
func nameFieldFor(meta *models.ObjectMetadata) string {
	for i := range meta.Fields {
		if meta.Fields[i].IsNameField {
			return meta.Fields[i].APIName
		}
	}
	return ""
}

// fanOutRecycleBinVisibility runs the per-object visibility queries
// concurrently up to recycleBinFanoutParallelism and aggregates the
// results. Per-object failures are captured as RecycleBinFailure entries —
// they never abort the overall fan-out, matching the design's
// partial-success contract.
func (ps *PersistenceService) fanOutRecycleBinVisibility(ctx context.Context, user *models.UserSession, scope RecycleBinScope, groups []recycleBinObjectGroup) ([]RecycleBinItem, []RecycleBinFailure, error) {
	type groupResult struct {
		items   []RecycleBinItem
		failure *RecycleBinFailure
	}
	results := make([]groupResult, len(groups))
	var wg sync.WaitGroup
	sem := make(chan struct{}, recycleBinFanoutParallelism)

	for index, group := range groups {
		// Acquire before launching so the concurrency budget bounds both active
		// queries and goroutine count. A semaphore inside the goroutine would
		// still create one waiter per object (up to the candidate-scan bound).
		select {
		case sem <- struct{}{}:
		case <-ctx.Done():
			wg.Wait()
			return nil, nil, ctx.Err()
		}
		wg.Add(1)
		go func(index int, group recycleBinObjectGroup) {
			defer wg.Done()
			defer func() { <-sem }()

			rows, err := ps.queryVisibleRecycleBinRows(ctx, user, scope, group)
			if err != nil {
				results[index].failure = &RecycleBinFailure{
					ObjectAPIName: group.apiName,
					ErrorCode:     errorCodeFor(err),
					Message:       sanitizedFailureMessage(err),
				}
				return
			}
			results[index].items = rows
		}(index, group)
	}
	wg.Wait()
	if err := ctx.Err(); err != nil {
		// Cancellation and deadlines are complete-operation outcomes. Returning
		// the subset that happened to finish would make scheduling order part of
		// the API and could misclassify cancellation as partial success.
		return nil, nil, err
	}

	var items []RecycleBinItem
	var failures []RecycleBinFailure
	for _, result := range results {
		items = append(items, result.items...)
		if result.failure != nil {
			failures = append(failures, *result.failure)
		}
	}
	return items, failures, nil
}

// queryVisibleRecycleBinRows re-reads one object's tombstoned records by id
// through QueryService, then merges each surviving row with its tombstone:
// delete-event facts from the tombstone, record facts from the live row.
func (ps *PersistenceService) queryVisibleRecycleBinRows(ctx context.Context, user *models.UserSession, scope RecycleBinScope, group recycleBinObjectGroup) ([]RecycleBinItem, error) {
	if len(group.recordIDs) == 0 {
		return nil, nil
	}

	selectFields := []string{
		constants.FieldID,
		constants.FieldIsDeleted,
		constants.FieldCreatedDate,
		constants.FieldLastModifiedDate,
	}
	if group.nameFieldReadable && group.nameField != "" && !strings.EqualFold(group.nameField, constants.FieldID) {
		selectFields = append(selectFields, group.nameField)
	}

	// Id IN (tombstoned ids) AND IsDeleted = true. The IN value MUST be
	// []interface{} — that is the shape QueryRepository type-asserts for
	// the IN operator (anything else is fmt-stringified and comma-split).
	filter := queryspec.And(
		queryspec.FieldPredicate(constants.FieldIsDeleted, "=", true),
		queryspec.FieldPredicate(constants.FieldID, "IN", group.recordIDs),
	)

	spec := queryspec.QuerySpec{
		ObjectAPIName: group.apiName,
		SelectFields:  selectFields,
		Filter:        filter,
		// queryAll semantics: include IsDeleted=true rows that the default
		// SOQL query() path would otherwise strip out.
		ScanAll: true,
	}

	queryCtx := ctx
	if scope == RecycleBinScopeAll {
		queryCtx = constants.WithStandardActionInternalAccessScopes(ctx, recycleBinScopedReadActionID, []constants.StandardActionInternalAccessScope{{
			StandardActionID: recycleBinScopedReadActionID,
			ObjectAPIName:    group.apiName,
			Operation:        constants.PermRead,
			EnforceFLS:       true,
		}})
	}
	result, err := ps.querySvc.QueryWithSpec(queryCtx, spec, user)
	if err != nil {
		return nil, err
	}
	if result == nil {
		return nil, nil
	}

	out := make([]RecycleBinItem, 0, len(result.Records))
	for _, row := range result.Records {
		id := stringValue(row[constants.FieldID])
		tombstone, ok := group.tombstones[id]
		if !ok {
			// A visible IsDeleted row we did not ask for; the IN list is
			// derived from tombstones, so this should not happen. Skip
			// rather than emit a row with no delete-event authority.
			continue
		}
		name := ""
		if group.nameFieldReadable && group.nameField != "" {
			name = recordNameValue(row, group.nameField)
		}
		if name == "" && group.nameField == "" {
			// The object has no declared name field, so the live read could
			// never have returned one; the tombstone's delete-time snapshot
			// is the only available label. We deliberately do NOT fall back
			// to the snapshot when the object HAS a name field but the
			// FLS-gated live read omitted it — doing so would leak a field
			// value the caller is not permitted to read (#1738 Codex round 3).
			name = tombstone.recordName
		}
		out = append(out, RecycleBinItem{
			ID:            id,
			RecordID:      id,
			RecordName:    name,
			ObjectAPIName: group.apiName,
			// Delete-event authority (tombstone), not mutable live audit.
			DeletedBy:   tombstone.deletedBy,
			DeletedDate: tombstone.deletedDate,
			tombstoneID: tombstone.id,
			// Record facts from the FLS-gated live read (RFC3339 wire shape,
			// matching the standard query/retrieve surfaces — see wireTimestamp).
			CreatedDate:      wireTimestamp(row[constants.FieldCreatedDate]),
			LastModifiedDate: wireTimestamp(row[constants.FieldLastModifiedDate]),
		})
	}
	return out, nil
}

// recordNameValue picks the object's declared name field; falls back to
// the standard "Name" column; returns "" when neither is present.
func recordNameValue(row models.SObject, nameField string) string {
	if nameField != "" {
		if v := stringValue(row[nameField]); v != "" {
			return v
		}
	}
	return stringValue(row[constants.FieldName])
}

// sortTombstonesByDeletedDateDesc orders tombstones newest-deleted-first.
// DeletedDate is an RFC3339-style string written by the same code path, so
// lexical comparison matches chronological order.
func sortTombstonesByDeletedDateDesc(tombstones []recycleBinTombstone) {
	sort.SliceStable(tombstones, func(i, j int) bool {
		return tombstones[i].deletedDate > tombstones[j].deletedDate
	})
}

// sortItemsByDeletedDateDesc orders the aggregated items by
// (DeletedDate DESC, tombstoneID DESC) — the same authoritative keyset the
// `_System_RecycleBin` page query uses. The tombstone-Id tie-break is REQUIRED:
// DeletedDate is second-granularity and the fan-out appends rows in
// goroutine-completion order, so without it the post-merge 200-row cap would be
// nondeterministic for same-second deletes and could drop newer tombstones
// across refreshes.
func sortItemsByDeletedDateDesc(items []RecycleBinItem) {
	sort.SliceStable(items, func(i, j int) bool {
		if items[i].DeletedDate != items[j].DeletedDate {
			return items[i].DeletedDate > items[j].DeletedDate
		}
		return items[i].tombstoneID > items[j].tombstoneID
	})
}

// wireTimestamp renders a recycle-bin DATETIME column to the v63 wire shape:
// RFC3339 in UTC (e.g. "2026-05-31T05:29:34Z", the Salesforce-canonical form).
// DATETIME columns arrive as time.Time from the driver (parseTime=True), so they
// MUST go through here rather than stringValue, whose fmt.Stringer branch emits
// Go's default "2006-01-02 15:04:05 +0000 UTC" layout. A value already in string
// form (a wire-shaped test fixture) passes through.
//
// It forces UTC because the read seam does not yet normalize DateTime: the
// standard query/retrieve path json-marshals the raw time.Time (RFC3339Nano in
// the DSN's loc=Local), which coincides with this output only on a UTC-deployed
// server. The canonical fix — normalize DateTime->wire once in FromStorageRecord
// (which already normalizes Date, but not DateTime) — is a v63-contract-wide
// change, deferred.
func wireTimestamp(v interface{}) string {
	switch t := v.(type) {
	case time.Time:
		return t.UTC().Format(time.RFC3339)
	case *time.Time:
		if t == nil {
			return ""
		}
		return t.UTC().Format(time.RFC3339)
	case nil:
		return ""
	case string:
		return t
	default:
		return stringValue(v)
	}
}

// dbDateTime formats a DATETIME column value for binding back into a keyset
// predicate against that same column (the recycle-bin page cursor). It is the
// sibling of wireTimestamp but for the SQL side, and differs deliberately:
//   - layout is the MySQL/TiDB DATETIME literal "2006-01-02 15:04:05", which the
//     driver binds straight into "DeletedDate < ?" — NOT RFC3339 ("T…Z") and NOT
//     stringValue's "2006-01-02 15:04:05 +0000 UTC", either of which mis-binds and
//     can skip/repeat/lose pages on the second tombstone page;
//   - it does NOT force UTC: the value is compared against the stored DATETIME,
//     which the driver reads in the DSN's loc, so the cursor must keep that same
//     wall clock rather than wireTimestamp's UTC wire shape.
func dbDateTime(v interface{}) string {
	switch t := v.(type) {
	case time.Time:
		return t.Format("2006-01-02 15:04:05")
	case *time.Time:
		if t == nil {
			return ""
		}
		return t.Format("2006-01-02 15:04:05")
	case nil:
		return ""
	case string:
		return t
	default:
		return stringValue(v)
	}
}

// stringValue is a tolerant value->string coercion for the non-timestamp
// recycle-bin SObject fields (record id, object api name, name, deleted-by).
// DATETIME columns arrive as time.Time (parseTime=True) and satisfy
// fmt.Stringer, so they must NOT be passed here — use wireTimestamp.
func stringValue(v interface{}) string {
	if v == nil {
		return ""
	}
	if s, ok := v.(string); ok {
		return s
	}
	if stringer, ok := v.(interface{ String() string }); ok {
		return stringer.String()
	}
	return fmt.Sprint(v)
}

// sanitizedFailureMessage returns a caller-safe message for the
// failedObjects[].message field. Only messages crafted as *apierr.Error
// are passed through verbatim — those are deliberate, user-facing
// strings the producer site already vetted for disclosure. Raw / wrapped
// errors (DB driver text, internal panics, infra strings) are collapsed
// to the empty string: the errorCode is the only machine-readable signal
// the client needs, and we don't want to leak SQL fragments, internal
// table identifiers, or upstream stack traces through this surface.
func sanitizedFailureMessage(err error) string {
	if err == nil {
		return ""
	}
	var apiErr *apierr.Error
	if errors.As(err, &apiErr) {
		return apiErr.Message
	}
	return ""
}

// errorCodeFor maps a service-layer error to a stable, machine-readable
// neutral code used in the partial-failure payload. API dialects own any
// carrier-specific wire vocabulary; the records engine only reports the
// class of failed per-object visibility read.
func errorCodeFor(err error) string {
	if err == nil {
		return string(contract.RecycleBinFailureCodeUnknown)
	}
	var statusErr interface{ HTTPStatus() int }
	if errors.As(err, &statusErr) {
		return string(recycleBinFailureCodeForHTTPStatus(statusErr.HTTPStatus()))
	}
	return string(contract.RecycleBinFailureCodeUnknown)
}

func recycleBinFailureCodeForHTTPStatus(status int) contract.RecycleBinFailureCode {
	switch status {
	case http.StatusBadRequest:
		return contract.RecycleBinFailureCodeInvalid
	case http.StatusForbidden, http.StatusUnauthorized:
		return contract.RecycleBinFailureCodeForbidden
	case http.StatusInternalServerError:
		return contract.RecycleBinFailureCodeInternal
	default:
		return contract.RecycleBinFailureCodeUnknown
	}
}