#!/bin/bash
# ==============================================================================================
# ================================= Play State Sync ============================================
# ==============================================================================================
#
# PURPOSE
# ─────────────────────────────────────────────────────────────────────────────
# Syncs watched/played state and resume positions across all configured Emby
# and Jellyfin servers. Newest timestamp wins — no data is ever lost.
#
# Users are matched by name (case-insensitive). If a user exists on some servers
# but not others, those servers are skipped for that user — no errors, no partial
# syncs from unrelated accounts.
#
# Items are matched by external provider IDs:
#   Movies   → IMDb ID, then TMDB ID
#   Episodes → TVDB ID + season + episode number
#   Audio    → MusicBrainz Track ID
#
# ==============================================================================================
# OPERATIONAL MODEL
# ==============================================================================================
#
# For each matched item across ≥2 servers:
#   1. Compare LastPlayedDate across all servers that have a play record.
#   2. The server with the newest LastPlayedDate is authoritative.
#   3. Push that server's state (Played, PlayCount, LastPlayedDate,
#      PlaybackPositionTicks) to every other server.
#   4. Servers with no record for that item also receive the state.
#
# Resume positions (partial plays, not marked Played):
#   Synced by comparing PlaybackPositionTicks when LastPlayedDate is absent.
#   The higher tick count wins.
#
# ==============================================================================================
# DESIGN PRINCIPLES
# ==============================================================================================
#
# Newest Timestamp Wins
#   No merge logic, no conflict resolution — the server with the most recent
#   LastPlayedDate is simply authoritative. Simple rules produce predictable
#   outcomes users can reason about.
#
# No Data Loss
#   The sync only pushes state forward — it never clears a Played flag or
#   resets a resume position to zero. A watch record on any server always
#   propagates outward, never disappears.
#
# Provider ID Matching
#   Items are matched by external IDs (IMDb, TVDB, MusicBrainz), not by
#   title or file path. This makes matching robust across library reorganisation,
#   renames, and multi-server path differences.
#
# ==============================================================================================
# OPERATIONAL SAFEGUARDS
# ==============================================================================================
#
# Root Enforcement
#   The probe fingerprint is written under STATE_DIR, which is not user-writable.
#   Without root the fingerprint silently fails to persist and the change probe
#   never suppresses anything.
#
# jq Dependency Check
#   Exits if jq is missing. All API response parsing and the epoch comparisons
#   depend on it — without jq every comparison would silently evaluate empty.
#
# PLAY_SYNC_ENABLED Gate
#   Exits cleanly when disabled; no partial runs.
#
# PLAY_SYNC_REMOTE Gate
#   When false, only this host's own servers are synced. Remote hosts are skipped
#   before any network call is attempted.
#
# Partnership Gate
#   Remote hosts are skipped when PARTNERSHIP_ENABLED=false. Local Emby↔Jellyfin
#   sync still runs — a dormant partnership does not disable local work.
#
# Tailscale Resolution Guard
#   A remote host whose Tailscale IP cannot be resolved is skipped rather than
#   contacted at its literal localhost URL, which would otherwise point the sync
#   at this host's own server and cross-contaminate state.
#
# Placeholder Credential Guard
#   Servers whose API key is empty or still a placeholder are dropped from the
#   list before any request is made.
#
# Per-Server Reachability
#   Unreachable servers are skipped individually; one offline server does not
#   abort the entire sync.
#
# User Match Required
#   A user missing from a server is skipped for that server. State is never
#   written to an unrelated account that happens to exist there.
#
# Forward-Only Writes
#   The sync only pushes state forward — it never clears a Played flag or resets
#   a resume position. The worst outcome of a bad comparison is a no-op, not
#   erased watch history.
#
# Lock Acquisition
#   acquire_lock prevents concurrent runs racing on the same items during the
#   30 minute critical window. --wait switches from skip to wait for manual runs.
#
# Probe Staleness Ceiling
#   PLAY_SYNC_PROBE_MAX_AGE_HOURS forces a full comparison regardless of the
#   hash. Fetches happen every run either way, so the probe can only skip
#   per-item processing — it can never cause a change to be missed outright.
#
# Dry Run Support
#   --dry-run performs all comparisons and writes no state.
#
# ==============================================================================================
# CONFIGURATION
# ==============================================================================================
#
# host*.conf
#
#   HOST*_TRANSCODE_SERVERS  "Name|URL|APIKey|type" entries per host (emby/jellyfin)
#                            Read directly for every HOST[0-9]+ defined — this script
#                            deliberately does NOT call detect_hosts(), because it needs
#                            every host's servers, not just this one's. Self is identified
#                            by comparing HOST* values against hostname -s.
#                            Remote host URLs have localhost rewritten to their Tailscale IP.
#
# master.conf
#
#   PLAY_SYNC_ENABLED        Master toggle (default: true)
#   PLAY_SYNC_REMOTE         Sync across all hosts via Tailscale (default: true)
#                            false = local servers only (this host's Emby + Jellyfin)
#   PLAY_SYNC_TYPES          Comma-separated item types to sync (default: Movie,Episode —
#                            Audio excluded, music library too large; favorites handled separately)
#   PLAY_SYNC_PROBE          Skip all per-item processing when no play/resume/favorite
#                            state changed since the last successful run (default: true).
#                            The raw API responses are hashed and compared against the
#                            fingerprint stored in STATE_DIR — fetches still happen every
#                            run, so nothing can be missed.
#   PLAY_SYNC_PROBE_MAX_AGE_HOURS  Force a full comparison when the stored fingerprint is
#                            older than this many hours regardless of the hash (default: 24)
#
# ==============================================================================================
# RUNTIME MODES
# ==============================================================================================
#
# play_state_sync.sh
#     Sync all matched users across all configured servers.
#
# play_state_sync.sh --dry-run
#     Show what would be synced without writing any state.
#
# play_state_sync.sh --status
#     Show configured servers, reachability, and user counts.
#
# play_state_sync.sh --full
#     Bypass the change probe — always run the full comparison.
#
# play_state_sync.sh --wait
#     Wait for an in-progress run to finish instead of exiting. For manual runs
#     that would otherwise be skipped by the every-30-minute scheduled pass.
#
# play_state_sync.sh --log
#     Verbose output — show each item comparison.
#
# ==============================================================================================

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
source "$SCRIPT_DIR/../load_config.sh"

# ── Handle --full and --wait before parse_args ────────────────────────────────
FULL_SYNC=false
LOCK_MODE="strict"
_FILTERED=()
for _a in "$@"; do
    if [[ "$_a" == "--full" ]]; then
        FULL_SYNC=true
    elif [[ "$_a" == "--wait" ]]; then
        LOCK_MODE="wait"
    else
        _FILTERED+=("$_a")
    fi
done
parse_args "${_FILTERED[@]}"

# ==============================================================================================
# ━━━ Setup ━━━
# ==============================================================================================
# The probe fingerprint lives under STATE_DIR — without root it silently fails to persist
# and the change probe can never suppress a run.
if [[ "$EUID" -ne 0 ]]; then
    error "Must be run as root"
    exit 1
fi

[[ "${PLAY_SYNC_ENABLED:-true}" != "true" ]] && echo "Play state sync disabled" && exit 0

SYNC_TYPES="${PLAY_SYNC_TYPES:-Movie,Episode}"
FAV_TYPES="${PLAY_SYNC_FAV_TYPES:-MusicArtist,MusicAlbum,Movie,Series}"
log "$ICON_GEAR Config: types=${SYNC_TYPES} favs=${FAV_TYPES} remote=${PLAY_SYNC_REMOTE:-true} probe=${PLAY_SYNC_PROBE:-true}"

command -v jq >/dev/null 2>&1 || { error "jq is required but not installed"; exit 1; }

acquire_lock "$LOCK_MODE"

# ── Build server list across ALL hosts ────────────────────────────────────────
# All HOST*_TRANSCODE_SERVERS arrays are loaded into env by load_config.sh.
# For remote hosts, localhost in the URL is rewritten to their Tailscale IP.
declare -a SRV_NAME SRV_URL SRV_KEY SRV_TYPE
_srv_count=0
_my_hostname=$(hostname -s)

_add_server() {
    local name="$1" url="$2" key="$3" type="$4"
    [[ -z "$url" || -z "$key" ]] && return
    [[ "$key" == "YOUR_API_KEY"* || "$key" == "placeholder"* ]] && return
    SRV_NAME[$_srv_count]="$name"
    SRV_URL[$_srv_count]="$url"
    SRV_KEY[$_srv_count]="$key"
    SRV_TYPE[$_srv_count]="$type"
    (( _srv_count++ ))
}

for _varname in $(compgen -v | grep -E '^HOST[0-9]+$' | sort); do
    _num="${_varname//[^0-9]/}"
    _host_hostname="${!_varname}"
    [[ -z "$_host_hostname" ]] && continue

    _is_me=false
    [[ "${_host_hostname,,}" == "${_my_hostname,,}" ]] && _is_me=true

    # Skip remote hosts when PLAY_SYNC_REMOTE=false
    if [[ "$_is_me" == false && "${PLAY_SYNC_REMOTE:-true}" != "true" ]]; then
        log "$_host_hostname — remote sync disabled (PLAY_SYNC_REMOTE=false), skipping"
        continue
    fi

    # Skip remote hosts when partnership is inactive
    if [[ "$_is_me" == false && "${PARTNERSHIP_ENABLED:-false}" != "true" ]]; then
        log "$_host_hostname — partnership inactive (PARTNERSHIP_ENABLED=false), skipping"
        continue
    fi

    # Resolve Tailscale IP for remote hosts
    _ts_ip=""
    if [[ "$_is_me" == false ]]; then
        _ts_ip=$(resolve_tailscale_ip "$_host_hostname")
        if [[ -z "$_ts_ip" ]]; then
            log "$_host_hostname — Tailscale IP not found, skipping"
            continue
        fi
    fi

    # Load this host's TRANSCODE_SERVERS array
    _srv_arr_name="HOST${_num}_TRANSCODE_SERVERS"
    eval "_host_entries=(\"\${${_srv_arr_name}[@]}\")"
    [[ "${#_host_entries[@]}" -eq 0 ]] && continue

    for _entry in "${_host_entries[@]}"; do
        IFS='|' read -r _name _url _key _type <<< "$_entry"
        [[ "$_type" == "emby" || "$_type" == "jellyfin" ]] || continue
        # For remote hosts rewrite localhost/127.0.0.1 → Tailscale IP
        if [[ "$_is_me" == false ]]; then
            _url="${_url//localhost/$_ts_ip}"
            _url="${_url//127.0.0.1/$_ts_ip}"
        fi
        _add_server "${_name} (${_host_hostname})" "$_url" "$_key" "$_type"
    done
done

if [[ "$_srv_count" -lt 2 ]]; then
    error "Need at least 2 media servers configured — found $_srv_count"
    exit 1
fi

# ==============================================================================================
# ━━━ API Helpers ━━━
# ==============================================================================================
_api_get() {
    local url="$1" key="$2" endpoint="$3"
    curl -sf --max-time 30 \
        -H "X-Emby-Token: $key" \
        "${url%/}/${endpoint}" 2>/dev/null
}

_api_post() {
    local url="$1" key="$2" endpoint="$3" data="${4:-}"
    if [[ -n "$data" ]]; then
        curl -sf --max-time 30 -s -o /dev/null -w "%{http_code}" -X POST \
            -H "X-Emby-Token: $key" \
            -H "Content-Type: application/json" \
            -d "$data" \
            "${url%/}/${endpoint}" 2>/dev/null
    else
        curl -sf --max-time 30 -s -o /dev/null -w "%{http_code}" -X POST \
            -H "X-Emby-Token: $key" \
            "${url%/}/${endpoint}" 2>/dev/null
    fi
}

# Ticks → seconds (1 tick = 100ns, 10_000_000 ticks = 1s)
_ticks_to_sec() {
    echo $(( ${1:-0} / 10000000 ))
}

# ==============================================================================================
# ━━━ Status ━━━
# ==============================================================================================
if [[ "$SHOW_STATUS" == true ]]; then
    echo ""
    echo "━━━━━ $ICON_SUMMARY PLAY STATE SYNC STATUS ━━━━━"
    echo "$ICON_GEAR  Remote sync: ${PLAY_SYNC_REMOTE:-true}"
    echo "$ICON_GEAR  Change probe: ${PLAY_SYNC_PROBE:-true}"
    echo "$ICON_GEAR  Item types:  $SYNC_TYPES"
    echo ""
    for i in $(seq 0 $(( _srv_count - 1 ))); do
        echo "$ICON_HOST  [${SRV_TYPE[$i]}] ${SRV_NAME[$i]}  (${SRV_URL[$i]})"
        _users=$(_api_get "${SRV_URL[$i]}" "${SRV_KEY[$i]}" "Users" 2>/dev/null | jq -r '.[].Name' 2>/dev/null | wc -l)
        if [[ "$_users" -gt 0 ]]; then
            echo "    $ICON_DONE Reachable — $_users user(s)"
            _api_get "${SRV_URL[$i]}" "${SRV_KEY[$i]}" "Users" 2>/dev/null \
                | jq -r '.[].Name' 2>/dev/null | while read -r n; do echo "      · $n"; done
        else
            echo "    $ICON_ERROR Unreachable or no users"
        fi
    done
    echo "━━━━━━━━━━━━━━━━━━━━━━━"
    exit 0
fi

# ==============================================================================================
# ━━━ Main Sync ━━━
# ==============================================================================================
echo "━━━ $ICON_SYNC Play State Sync — $(date '+%Y-%m-%d %H:%M:%S') ━━━"
[[ "$DRY_RUN" == true ]]  && warn "DRY RUN — no state will be written"
[[ "$FULL_SYNC" == true ]] && log "Full sync mode — change probe bypassed"

START=$(date +%s)
TOTAL_SYNCED=0
TOTAL_SKIPPED=0
TOTAL_ERRORS=0

# ── Step 1: Fetch users from each server ─────────────────────────────────────
declare -A SRV_USERS   # idx → JSON array string of users

log "Fetching users..."
for i in $(seq 0 $(( _srv_count - 1 ))); do
    _resp=$(_api_get "${SRV_URL[$i]}" "${SRV_KEY[$i]}" "Users")
    if [[ -z "$_resp" ]]; then
        warn "${SRV_NAME[$i]} — unreachable, skipping"
        SRV_USERS[$i]=""
        continue
    fi
    SRV_USERS[$i]="$_resp"
    _count=$(echo "$_resp" | jq 'length' 2>/dev/null || echo 0)
    log "${SRV_NAME[$i]} — $_count user(s)"
done

# ── Step 2: Build cross-server user map ──────────────────────────────────────
# lowercase_name → "server_idx:user_id server_idx:user_id ..."
declare -A USER_MAP

for i in $(seq 0 $(( _srv_count - 1 ))); do
    [[ -z "${SRV_USERS[$i]}" ]] && continue
    while IFS=$'\t' read -r uid uname; do
        [[ -z "$uid" || -z "$uname" ]] && continue
        lname="${uname,,}"
        if [[ -n "${USER_MAP[$lname]}" ]]; then
            USER_MAP[$lname]+=" ${i}:${uid}"
        else
            USER_MAP[$lname]="${i}:${uid}"
        fi
    done < <(echo "${SRV_USERS[$i]}" | jq -r '.[] | [.Id, .Name] | @tsv' 2>/dev/null)
done

# ── Step 2.3: Fetch play/resume/favorite state for every matched user ────────
# Raw responses are cached for the sync passes below and hashed for the change
# probe. Fetching costs seconds — the per-item comparison is what costs minutes,
# so it only runs when a response actually changed since the last successful run.
declare -A RESP_STATE   # "lname|si" → deduplicated played+resumable items JSON
declare -A RESP_FAV     # "lname|si|ftype" → favorites response JSON

IFS=',' read -ra _fav_type_list <<< "$FAV_TYPES"

for lname in "${!USER_MAP[@]}"; do
    read -ra _pairs <<< "${USER_MAP[$lname]}"
    [[ "${#_pairs[@]}" -lt 2 ]] && continue

    for _pair in "${_pairs[@]}"; do
        IFS=':' read -r _si _uid <<< "$_pair"

        # All played items — no limit, covers both date-stamped and batch-marked (null date) entries
        _endpoint="Users/${_uid}/Items?Recursive=true&Fields=ProviderIds,UserData,Type,ParentIndexNumber,IndexNumber,SeriesName&IncludeItemTypes=${SYNC_TYPES}&Filters=IsPlayed&SortBy=DatePlayed&SortOrder=Descending"
        _resp=$(_api_get "${SRV_URL[$_si]}" "${SRV_KEY[$_si]}" "$_endpoint")
        if [[ -z "$_resp" ]]; then
            warn "  ${SRV_NAME[$_si]} — failed to fetch items for $lname"
            RESP_STATE["$lname|$_si"]=""
        else
            # Resume positions (not yet marked played)
            _endpoint2="Users/${_uid}/Items?Recursive=true&Fields=ProviderIds,UserData,Type,ParentIndexNumber,IndexNumber,SeriesName&IncludeItemTypes=${SYNC_TYPES}&SortBy=DatePlayed&SortOrder=Descending&Filters=IsResumable"
            _resp2=$(_api_get "${SRV_URL[$_si]}" "${SRV_KEY[$_si]}" "$_endpoint2")

            # Combine and deduplicate by Id
            [[ -z "$_resp2" ]] && _resp2='{"Items":[]}'
            RESP_STATE["$lname|$_si"]=$(printf '%s\n%s' "$_resp" "$_resp2" | jq -s \
                '[.[0].Items // [], .[1].Items // []] | add // [] | unique_by(.Id)' 2>/dev/null)
        fi

        for _ftype in "${_fav_type_list[@]}"; do
            RESP_FAV["$lname|$_si|$_ftype"]=$(_api_get "${SRV_URL[$_si]}" "${SRV_KEY[$_si]}" \
                "Users/${_uid}/Items?Recursive=true&IncludeItemTypes=${_ftype}&Filters=IsFavorite&Fields=ProviderIds")
        done
    done
done

# ── Step 2.4: Change probe — skip the comparison when nothing changed ────────
# Failed fetches hash as empty strings, so reachability transitions also read
# as changes and trigger a full pass once the server comes back.
PROBE_FILE="$STATE_DIR/play_state_sync_probe"
CUR_HASH=$(
    {
        while IFS= read -r _k; do
            printf '%s:' "$_k"; printf '%s' "${RESP_STATE[$_k]}" | md5sum
        done < <(printf '%s\n' "${!RESP_STATE[@]}" | sort)
        while IFS= read -r _k; do
            printf '%s:' "$_k"; printf '%s' "${RESP_FAV[$_k]}" | md5sum
        done < <(printf '%s\n' "${!RESP_FAV[@]}" | sort)
    } | md5sum | awk '{print $1}'
)

if [[ "${PLAY_SYNC_PROBE:-true}" == "true" && "$FULL_SYNC" == false && "$DRY_RUN" == false && -f "$PROBE_FILE" ]]; then
    _prev_hash=$(sed -n '1p' "$PROBE_FILE" 2>/dev/null)
    _prev_epoch=$(sed -n '2p' "$PROBE_FILE" 2>/dev/null)
    [[ "$_prev_epoch" =~ ^[0-9]+$ ]] || _prev_epoch=0
    _probe_max_age=$(( ${PLAY_SYNC_PROBE_MAX_AGE_HOURS:-24} * 3600 ))
    if [[ "$CUR_HASH" == "$_prev_hash" ]] && (( $(date +%s) - ${_prev_epoch:-0} < _probe_max_age )); then
        END=$(date +%s)
        echo ""
        echo "━━━━━ $ICON_SUMMARY PLAY STATE SYNC SUMMARY ━━━━━"
        echo "$ICON_TIME  Duration:  $(format_duration $(( END - START )))"
        echo "$ICON_DONE  No play/resume/favorite changes since last sync — comparison skipped"
        success "Done ✅"
        exit 0
    fi
fi

# ── Step 2.5: Pre-build provider ID → item ID lookup map ─────────────────────
# AnyProviderIdEquals is broken in Jellyfin 10.11+ (ignores the filter entirely).
# Pre-fetching all items' provider IDs once and using a local map avoids the broken
# per-item API call and is faster overall.
declare -A PROV_LOOKUP   # "si|tvdb.{id}" | "si|imdb.{id}" | "si|mb.{id}" → item_id

# Jellyfin's mixed-type query (Movie,Episode,Audio) changes sort order unpredictably,
# pushing items to positions far beyond the first pages. Query each type separately so
# each list sorts within its own type and items appear where expected.
_PROV_PAGE=2000
# Combine play state types + fav types for the map build (deduplicated)
declare -A _prov_seen_dedup
_prov_types=()
IFS=',' read -ra _prov_combined <<< "${SYNC_TYPES},${FAV_TYPES}"
for _t in "${_prov_combined[@]}"; do
    [[ -n "${_prov_seen_dedup[$_t]:-}" ]] && continue
    _prov_seen_dedup[$_t]=1; _prov_types+=("$_t")
done
unset _prov_seen_dedup _prov_combined

for _psi in $(seq 0 $(( _srv_count - 1 ))); do
    [[ -z "${SRV_URL[$_psi]}" ]] && continue
    log "Building provider ID map for ${SRV_NAME[$_psi]}..."
    for _prov_type in "${_prov_types[@]}"; do
        _prov_start=0
        _prov_total=-1
        while true; do
            _page=$(curl -sf --max-time 60 \
                -H "X-Emby-Token: ${SRV_KEY[$_psi]}" \
                "${SRV_URL[$_psi]%/}/Items?Recursive=true&IncludeItemTypes=${_prov_type}&Fields=ProviderIds&Limit=${_PROV_PAGE}&StartIndex=${_prov_start}" 2>/dev/null)
            [[ -z "$_page" ]] && break
            [[ "$_prov_total" -lt 0 ]] && _prov_total=$(echo "$_page" | jq '.TotalRecordCount // 0' 2>/dev/null || echo 0)
            _prov_count=$(echo "$_page" | jq '.Items | length' 2>/dev/null || echo 0)
            [[ "$_prov_count" -eq 0 ]] && break
            while IFS=$'\t' read -r _pid _ptype _ptvdb _pimdb _ptmdb _pmbtrack _pmbartist _pmbalbum; do
                [[ "$_pimdb"     != "null" && -n "$_pimdb"     ]] && PROV_LOOKUP["${_psi}|imdb.${_pimdb}"]="$_pid"
                [[ "$_ptmdb"     != "null" && -n "$_ptmdb"     ]] && PROV_LOOKUP["${_psi}|tmdb.${_ptmdb}"]="$_pid"
                [[ "$_pmbtrack"  != "null" && -n "$_pmbtrack"  ]] && PROV_LOOKUP["${_psi}|mb.${_pmbtrack}"]="$_pid"
                [[ "$_pmbartist" != "null" && -n "$_pmbartist" ]] && PROV_LOOKUP["${_psi}|mb.artist.${_pmbartist}"]="$_pid"
                [[ "$_pmbalbum"  != "null" && -n "$_pmbalbum"  ]] && PROV_LOOKUP["${_psi}|mb.album.${_pmbalbum}"]="$_pid"
                # tvdb: namespace by type so Series and Episode IDs don't collide
                if [[ "$_ptvdb" != "null" && -n "$_ptvdb" ]]; then
                    case "$_ptype" in
                        Series)  PROV_LOOKUP["${_psi}|tvdb.series.${_ptvdb}"]="$_pid" ;;
                        Episode) PROV_LOOKUP["${_psi}|tvdb.${_ptvdb}"]="$_pid" ;;
                        *)       PROV_LOOKUP["${_psi}|tvdb.${_ptvdb}"]="$_pid" ;;
                    esac
                fi
            done < <(echo "$_page" | jq -r '.Items[] | [
                .Id,
                .Type,
                (.ProviderIds.Tvdb // "null"),
                (.ProviderIds.Imdb // "null"),
                (.ProviderIds.Tmdb // "null"),
                (.ProviderIds.MusicBrainzTrackId // "null"),
                (.ProviderIds.MusicBrainzArtistId // "null"),
                (.ProviderIds.MusicBrainzAlbumId // "null")
            ] | @tsv' 2>/dev/null)
            _prov_start=$(( _prov_start + _prov_count ))
            [[ "$_prov_total" -gt 0 && "$_prov_start" -ge "$_prov_total" ]] && break
        done
    done
done

# ── Step 3: Sync per matched user ────────────────────────────────────────────
for lname in "${!USER_MAP[@]}"; do
    read -ra _pairs <<< "${USER_MAP[$lname]}"

    # Skip users only on one server
    [[ "${#_pairs[@]}" -lt 2 ]] && log "  $lname — only on 1 server, skipping" && continue

    echo ""
    echo "── User: $lname (${#_pairs[@]} server(s)) ──"

    # Build per-server user context
    declare -A U_IDX U_UID
    for _pair in "${_pairs[@]}"; do
        IFS=':' read -r _si _ui <<< "$_pair"
        U_IDX["$_si"]="$_si"
        U_UID["$_si"]="$_ui"
    done

    # ── Fetch played items from each server for this user ────────────────────
    # Key: provider_id_string → sorted list of (epoch, srv_idx, item_id, play_count, ticks, played)
    declare -A ITEM_MAP   # provider_key → JSON per-server data

    for _si in "${!U_IDX[@]}"; do
        _combined="${RESP_STATE[$lname|$_si]:-}"
        [[ -z "$_combined" ]] && continue

        _count=$(echo "$_combined" | jq 'length' 2>/dev/null || echo 0)
        log "  ${SRV_NAME[$_si]} — $_count item(s) with state for $lname"

        # Build item lookup by provider key
        while IFS=$'\t' read -r iid itype season ep imdb tmdb tvdb mbtrack played ticks lplayed pcount epoch; do
            # Build canonical provider key
            _pkey=""
            case "$itype" in
                Movie)
                    [[ "$imdb" != "null" && -n "$imdb" ]] && _pkey="imdb:${imdb}"
                    [[ -z "$_pkey" && "$tmdb" != "null" && -n "$tmdb" ]] && _pkey="tmdb:movie:${tmdb}"
                    ;;
                Episode)
                    [[ "$tvdb" != "null" && -n "$tvdb" && "$season" != "null" && "$ep" != "null" ]] && \
                        _pkey="tvdb:ep:${tvdb}:s${season}e${ep}"
                    ;;
                Audio)
                    [[ "$mbtrack" != "null" && -n "$mbtrack" ]] && _pkey="mb:track:${mbtrack}"
                    ;;
            esac
            [[ -z "$_pkey" ]] && continue

            _epoch="${epoch:-0}"
            _entry="${_si}|${iid}|${played}|${pcount}|${ticks}|${_epoch}|${lplayed}"

            if [[ -n "${ITEM_MAP[$_pkey]}" ]]; then
                ITEM_MAP[$_pkey]+=$'\n'"$_entry"
            else
                ITEM_MAP[$_pkey]="$_entry"
            fi

        done < <(echo "$_combined" | jq -r '.[] | [
            .Id,
            .Type,
            (.ParentIndexNumber // "null" | tostring),
            (.IndexNumber // "null" | tostring),
            (.ProviderIds.Imdb // "null"),
            (.ProviderIds.Tmdb // "null"),
            (.ProviderIds.Tvdb // "null"),
            (.ProviderIds.MusicBrainzTrackId // "null"),
            (.UserData.Played // false | tostring),
            (.UserData.PlaybackPositionTicks // 0 | tostring),
            (.UserData.LastPlayedDate // "null"),
            (.UserData.PlayCount // 0 | tostring),
            ((.UserData.LastPlayedDate // null) | if . == null then "0"
             else (((sub("\\.[0-9]+"; "") | sub("[+-][0-9]{2}:?[0-9]{2}$"; "Z")
                     | if endswith("Z") then . else . + "Z" end
                     | fromdateiso8601)? // 0) | tostring) end)
        ] | @tsv' 2>/dev/null)
    done

    # ── Compare and sync ─────────────────────────────────────────────────────
    for _pkey in "${!ITEM_MAP[@]}"; do
        # Collect all server entries for this item
        declare -A E_EPOCH E_PLAYED E_PCOUNT E_TICKS E_IID E_LPLAYED E_SIDX
        _has_entries=false

        while IFS='|' read -r _si _iid _played _pcount _ticks _epoch _lplayed; do
            [[ -z "$_si" ]] && continue
            E_SIDX[$_si]="$_si"
            E_IID[$_si]="$_iid"
            E_PLAYED[$_si]="$_played"
            E_PCOUNT[$_si]="$_pcount"
            E_TICKS[$_si]="$_ticks"
            E_EPOCH[$_si]="$_epoch"
            E_LPLAYED[$_si]="$_lplayed"
            _has_entries=true
        done <<< "${ITEM_MAP[$_pkey]}"

        [[ "$_has_entries" == false ]] && continue

        # Find the authoritative server: newest LastPlayedDate epoch
        # Tie-break: higher PlayCount, then higher Ticks, then Played=true
        # Init at -1 so servers with epoch=0 (batch-marks with null LastPlayedDate) can win
        _auth_si=""
        _auth_epoch=-1
        _auth_pcount=-1
        _auth_ticks=-1
        _auth_pf="false"

        for _si in "${!E_SIDX[@]}"; do
            _e="${E_EPOCH[$_si]:-0}"
            _pc="${E_PCOUNT[$_si]:-0}"
            _tk="${E_TICKS[$_si]:-0}"
            _pf="${E_PLAYED[$_si]:-false}"
            # Played=true is the primary key — a played server always beats a non-played server
            # regardless of epoch. A resumable item with a newer LastPlayedDate must not become
            # authority over a played item, as pushing resume ticks to a played server resets
            # the played status on some Emby/Jellyfin versions.
            if ( [[ "$_pf" == "true" ]] && [[ "$_auth_pf" != "true" ]] ) || \
               ( [[ "$_pf" == "$_auth_pf" ]] && [[ "$_e" -gt "$_auth_epoch" ]] ) || \
               ( [[ "$_pf" == "$_auth_pf" ]] && [[ "$_e" -eq "$_auth_epoch" ]] && [[ "$_pc" -gt "$_auth_pcount" ]] ) || \
               ( [[ "$_pf" == "$_auth_pf" ]] && [[ "$_e" -eq "$_auth_epoch" ]] && [[ "$_pc" -eq "$_auth_pcount" ]] && [[ "$_tk" -gt "$_auth_ticks" ]] ); then
                _auth_si="$_si"
                _auth_epoch="$_e"
                _auth_pcount="$_pc"
                _auth_ticks="$_tk"
                _auth_pf="$_pf"
            fi
        done

        [[ -z "$_auth_si" ]] && continue

        _auth_played="${E_PLAYED[$_auth_si]}"
        _auth_lplayed="${E_LPLAYED[$_auth_si]}"
        _auth_pcount="${E_PCOUNT[$_auth_si]}"
        _auth_ticks="${E_TICKS[$_auth_si]}"

        # Push to servers with older state OR no state at all
        for _si in "${!U_IDX[@]}"; do
            [[ "$_si" == "$_auth_si" ]] && continue
            _their_epoch="${E_EPOCH[$_si]:-0}"
            _their_played="${E_PLAYED[$_si]:-false}"

            # Skip if they already have the same/newer state
            _skip=false
            if [[ "$_auth_played" == "true" ]]; then
                # Both servers already have this played — nothing to propagate regardless of dates.
                # Date-based comparison caused a ping-pong: syncing without a DatePlayed param lets
                # the target server stamp the current time, making it the new authority next cycle.
                [[ "$_their_played" == "true" ]] && _skip=true
            else
                # Resume only: skip if target already has same or more ticks.
                # Allow 5-second tolerance (50_000_000 ticks) — Emby may round tick values slightly
                # differently on read, causing an exact-match check to miss and re-sync every cycle.
                _tick_gap=$(( ${_auth_ticks:-0} - ${E_TICKS[$_si]:-0} ))
                [[ "$_tick_gap" -le 50000000 && "${E_TICKS[$_si]:-0}" -gt 0 && "$_their_played" == "false" ]] && _skip=true
            fi
            if [[ "$_skip" == true ]]; then
                log "  SKIP  $_pkey → ${SRV_NAME[$_si]} already up to date"
                (( TOTAL_SKIPPED++ ))
                continue
            fi

            _uid="${U_UID[$_si]}"
            _iid="${E_IID[$_si]:-}"   # might not exist on this server yet

            # Find item ID on target server by provider key if not in our map
            if [[ -z "$_iid" ]]; then
                _ptype="${_pkey%%:*}"
                case "$_ptype" in
                    imdb)
                        _pval="${_pkey#imdb:}"
                        _iid="${PROV_LOOKUP[${_si}|imdb.${_pval}]:-}"
                        ;;
                    tmdb)
                        _pval="${_pkey#tmdb:movie:}"
                        _iid="${PROV_LOOKUP[${_si}|tmdb.${_pval}]:-}"
                        ;;
                    tvdb)
                        # pkey format: tvdb:ep:{tvdb_id}:s{season}e{ep}
                        _tvdb_num="${_pkey#tvdb:ep:}"; _tvdb_num="${_tvdb_num%%:*}"
                        _iid="${PROV_LOOKUP[${_si}|tvdb.${_tvdb_num}]:-}"
                        ;;
                    mb)
                        _pval="${_pkey#mb:track:}"
                        _iid="${PROV_LOOKUP[${_si}|mb.${_pval}]:-}"
                        ;;
                esac
                [[ -z "$_iid" ]] && log "  SKIP  $_pkey → ${SRV_NAME[$_si]} item not found on server" && continue
            fi

            log "  SYNC  $_pkey → ${SRV_NAME[$_si]} (auth: ${SRV_NAME[$_auth_si]}, epoch: $_auth_epoch)"

            if [[ "$DRY_RUN" == true ]]; then
                echo "  DRY RUN: would sync $_pkey → ${SRV_NAME[$_si]} user=$lname played=$_auth_played date=$_auth_lplayed"
                (( TOTAL_SYNCED++ ))
                continue
            fi

            # Write state to target server
            if [[ "$_auth_played" == "true" ]]; then
                # Mark as played with date
                _date_param=""
                if [[ "$_auth_lplayed" != "null" && -n "$_auth_lplayed" ]]; then
                    # PlayedItems expects DatePlayed in yyyyMMddHHmmss (no separators, no timezone)
                    _lp="${_auth_lplayed%.*}"
                    _lp="${_lp%Z}"
                    _date_param="?DatePlayed=${_lp//[-T:]/}"
                fi
                _http=$(_api_post "${SRV_URL[$_si]}" "${SRV_KEY[$_si]}" \
                    "Users/${_uid}/PlayedItems/${_iid}${_date_param}")
                if [[ "$_http" == "200" || "$_http" == "201" ]]; then
                    success "  ✓  $_pkey → ${SRV_NAME[$_si]} marked played"
                    (( TOTAL_SYNCED++ ))
                else
                    warn "  ✗  $_pkey → ${SRV_NAME[$_si]} failed (HTTP ${_http:-err})"
                    (( TOTAL_ERRORS++ ))
                fi
            else
                # Sync resume position only — never push ticks to a server that already
                # has this item marked played. Writing ticks via UserData can reset Played=false.
                if [[ "${E_PLAYED[$_si]:-false}" == "true" ]]; then
                    log "  SKIP  $_pkey → ${SRV_NAME[$_si]} already played, not overwriting with resume ticks"
                    (( TOTAL_SKIPPED++ ))
                    continue
                fi
                _ticks_int=$(( ${_auth_ticks:-0} ))
                if [[ "$_ticks_int" -gt 0 ]]; then
                    _payload="{\"PlaybackPositionTicks\":${_ticks_int}}"
                    _http=$(_api_post "${SRV_URL[$_si]}" "${SRV_KEY[$_si]}" \
                        "Users/${_uid}/Items/${_iid}/UserData" "$_payload")
                    if [[ "$_http" == "200" || "$_http" == "204" ]]; then
                        success "  ✓  $_pkey → ${SRV_NAME[$_si]} resume synced ($(_ticks_to_sec "$_ticks_int")s)"
                        (( TOTAL_SYNCED++ ))
                    else
                        warn "  ✗  $_pkey → ${SRV_NAME[$_si]} resume sync failed (HTTP ${_http:-err})"
                        (( TOTAL_ERRORS++ ))
                    fi
                fi
            fi
        done

        unset E_EPOCH E_PLAYED E_PCOUNT E_TICKS E_IID E_LPLAYED E_SIDX
    done

    unset ITEM_MAP U_IDX U_UID
done

# ==============================================================================================
# ━━━ Favorite Sync ━━━
# ==============================================================================================
# Union semantics — if favorited on any server, sync to all others. Never unmarks.
# Covers MusicArtist, MusicAlbum, Movie, Series (configured via PLAY_SYNC_FAV_TYPES).
FAV_TOTAL_SYNCED=0
FAV_TOTAL_SKIPPED=0
FAV_TOTAL_ERRORS=0

for lname in "${!USER_MAP[@]}"; do
    read -ra _pairs <<< "${USER_MAP[$lname]}"
    [[ "${#_pairs[@]}" -lt 2 ]] && continue

    declare -A UF_IDX UF_UID
    for _pair in "${_pairs[@]}"; do
        IFS=':' read -r _si _ui <<< "$_pair"
        UF_IDX["$_si"]="$_si"
        UF_UID["$_si"]="$_ui"
    done

    declare -A FAV_MAP   # provider_key → space-separated "si:iid" pairs

    for _si in "${!UF_IDX[@]}"; do
        _uid="${UF_UID[$_si]}"
        for _ftype in "${_fav_type_list[@]}"; do
            _resp="${RESP_FAV[$lname|$_si|$_ftype]:-}"
            [[ -z "$_resp" ]] && continue
            _fcount=$(echo "$_resp" | jq '.Items | length' 2>/dev/null || echo 0)
            [[ "$_fcount" -eq 0 ]] && continue
            log "  ${SRV_NAME[$_si]} — $_fcount ${_ftype} favorite(s) for $lname"

            while IFS=$'\t' read -r _iid _itype _pimdb _ptmdb _ptvdb _pmbartist _pmbalbum _pmbtrack; do
                _pkey=""
                case "$_itype" in
                    Movie)
                        [[ "$_pimdb" != "null" && -n "$_pimdb" ]] && _pkey="imdb:${_pimdb}"
                        [[ -z "$_pkey" && "$_ptmdb" != "null" && -n "$_ptmdb" ]] && _pkey="tmdb:movie:${_ptmdb}"
                        ;;
                    Series)
                        [[ "$_ptvdb" != "null" && -n "$_ptvdb" ]] && _pkey="tvdb:series:${_ptvdb}"
                        ;;
                    MusicArtist)
                        [[ "$_pmbartist" != "null" && -n "$_pmbartist" ]] && _pkey="mb:artist:${_pmbartist}"
                        ;;
                    MusicAlbum)
                        [[ "$_pmbalbum" != "null" && -n "$_pmbalbum" ]] && _pkey="mb:album:${_pmbalbum}"
                        ;;
                    Audio)
                        [[ "$_pmbtrack" != "null" && -n "$_pmbtrack" ]] && _pkey="mb:track:${_pmbtrack}"
                        ;;
                esac
                [[ -z "$_pkey" ]] && continue

                if [[ -n "${FAV_MAP[$_pkey]:-}" ]]; then
                    FAV_MAP[$_pkey]+=" ${_si}:${_iid}"
                else
                    FAV_MAP[$_pkey]="${_si}:${_iid}"
                fi
            done < <(echo "$_resp" | jq -r '.Items[] | [
                .Id, .Type,
                (.ProviderIds.Imdb // "null"),
                (.ProviderIds.Tmdb // "null"),
                (.ProviderIds.Tvdb // "null"),
                (.ProviderIds.MusicBrainzArtistId // "null"),
                (.ProviderIds.MusicBrainzAlbumId // "null"),
                (.ProviderIds.MusicBrainzTrackId // "null")
            ] | @tsv' 2>/dev/null)
        done
    done

    for _pkey in "${!FAV_MAP[@]}"; do
        declare -A _fhave _fiid
        for _entry in ${FAV_MAP[$_pkey]}; do
            IFS=':' read -r _si _iid <<< "$_entry"
            _fhave[$_si]=true
            _fiid[$_si]="$_iid"
        done

        for _si in "${!UF_IDX[@]}"; do
            [[ "${_fhave[$_si]:-false}" == "true" ]] && (( FAV_TOTAL_SKIPPED++ )) && continue
            _uid="${UF_UID[$_si]}"

            # Look up item ID on target server via PROV_LOOKUP
            _iid=""
            _ptype="${_pkey%%:*}"
            case "$_ptype" in
                imdb)  _iid="${PROV_LOOKUP[${_si}|imdb.${_pkey#imdb:}]:-}" ;;
                tmdb)  _iid="${PROV_LOOKUP[${_si}|tmdb.${_pkey#tmdb:movie:}]:-}" ;;
                tvdb)  _iid="${PROV_LOOKUP[${_si}|tvdb.series.${_pkey#tvdb:series:}]:-}" ;;
                mb)
                    _mbsub="${_pkey#mb:}"
                    case "${_mbsub%%:*}" in
                        artist) _iid="${PROV_LOOKUP[${_si}|mb.artist.${_mbsub#artist:}]:-}" ;;
                        album)  _iid="${PROV_LOOKUP[${_si}|mb.album.${_mbsub#album:}]:-}" ;;
                        track)  _iid="${PROV_LOOKUP[${_si}|mb.${_mbsub#track:}]:-}" ;;
                    esac
                    ;;
            esac

            if [[ -z "$_iid" ]]; then
                log "  SKIP  FAV $_pkey → ${SRV_NAME[$_si]} not in library"
                continue
            fi

            log "  FAV   $_pkey → ${SRV_NAME[$_si]} ($lname)"
            if [[ "$DRY_RUN" == false ]]; then
                _http=$(_api_post "${SRV_URL[$_si]}" "${SRV_KEY[$_si]}" \
                    "Users/${_uid}/FavoriteItems/${_iid}")
                if [[ "$_http" == "200" || "$_http" == "201" ]]; then
                    success "  ✓  FAV $_pkey → ${SRV_NAME[$_si]}"
                    (( FAV_TOTAL_SYNCED++ ))
                else
                    warn "  ✗  FAV $_pkey → ${SRV_NAME[$_si]} failed (HTTP ${_http:-err})"
                    (( FAV_TOTAL_ERRORS++ ))
                fi
            else
                echo "  DRY RUN: would favorite $_pkey → ${SRV_NAME[$_si]} user=$lname"
                (( FAV_TOTAL_SYNCED++ ))
            fi
        done

        unset _fhave _fiid
    done

    unset FAV_MAP UF_IDX UF_UID
done

# ==============================================================================================
# ━━━ Summary ━━━
# ==============================================================================================
END=$(date +%s)
echo ""
echo "━━━━━ $ICON_SUMMARY PLAY STATE SYNC SUMMARY ━━━━━"
echo "$ICON_TIME  Duration:  $(format_duration $(( END - START )))"
echo "$ICON_DONE  Synced:    $TOTAL_SYNCED"
[[ "$TOTAL_SKIPPED" -gt 0 ]] && echo "$ICON_RUNNING  Skipped:   $TOTAL_SKIPPED (already current)"
[[ "$TOTAL_ERRORS"  -gt 0 ]] && echo "$ICON_ERROR  Errors:    $TOTAL_ERRORS"
echo ""
echo "  Favorites — synced: $FAV_TOTAL_SYNCED  skipped: $FAV_TOTAL_SKIPPED  errors: $FAV_TOTAL_ERRORS"

TOTAL_ALL_ERRORS=$(( TOTAL_ERRORS + FAV_TOTAL_ERRORS ))

# Fingerprint is the PRE-sync state — our own writes above changed the targets,
# so the next run does one more full pass and then settles into probe skips.
# Re-hashing post-sync instead would swallow plays that landed mid-run.
if [[ "$DRY_RUN" == false && "$TOTAL_ALL_ERRORS" -eq 0 ]]; then
    printf '%s\n%s\n' "$CUR_HASH" "$START" > "$PROBE_FILE" 2>/dev/null || true
fi

if [[ "$DRY_RUN" == true ]]; then
    warn "DRY RUN — no changes written"
    exit 0
elif [[ "$TOTAL_ALL_ERRORS" -eq 0 ]]; then
    success "Done ✅"
    exit 0
else
    warn "Done with $TOTAL_ALL_ERRORS error(s)"
    exit 1
fi