#!/usr/bin/env bash
# YMServer isolated DB only. Unique vessel, one-shot, does not clean other tests.
set -euo pipefail
ROOT="$(cd "$(dirname "$0")/.." && pwd)"
cd "$ROOT"
DB=yachtmemory_catalog_v2_test
VESSEL=a20b89f5-1fd9-42af-8919-7db720aa8c31
SID=b2fa114c-48ca-48b5-8096-32a9b5f316be
R="$ROOT/receiver/ym-sync-catalog-v2-atomic-review.php"
[[ $(id -un) != root ]] || { echo 'ABORT: run as standard YMServer user'; exit 1; }
[[ $(grep -Fxc "const DB_NAME = 'yachtmemory_catalog_v2_test';" "$R") == 1 ]] || { echo 'ABORT: receiver DB mismatch'; exit 1; }
php -l "$R"
q() { sudo mysql -N "$DB" -e "$1"; }
[[ $(sudo mysql -N -e "SELECT COUNT(*) FROM information_schema.SCHEMATA WHERE SCHEMA_NAME='$DB'") == 1 ]] || { echo 'ABORT: missing DB';exit 1; }
[[ $(sudo mysql -N -e "SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA='$DB' AND TABLE_NAME='sync_receipts'") == 1 ]] || { echo 'ABORT: missing receipts';exit 1; }
[[ $(q "SELECT COUNT(*) FROM telemetry_vessels WHERE vessel_id=UUID_TO_BIN('$VESSEL')") == 0 ]] || { echo 'ABORT: synthetic vessel already exists (one-shot)'; exit 1; }
[[ $(q "SELECT COUNT(*) FROM signalk_catalog_snapshots_v2 WHERE snapshot_id=UUID_TO_BIN('$SID')") == 0 ]] || { echo 'ABORT: synthetic snapshot exists'; exit 1; }
# No production tables touched, all SQL statements specify isolated DB.
sudo mysql "$DB" -e "INSERT INTO telemetry_vessels(vessel_id,vessel_key,created_at_utc) VALUES(UUID_TO_BIN('$VESSEL'),'catalog-v2-concurrency-011',UTC_TIMESTAMP(3))"
OUT=$(mktemp -d /tmp/ym-concurrency-011-XXXXXX)
echo "TEST LOG DIRECTORY: $OUT (retained for review)"
run() { sudo -n env -u YM_CATALOG_ATOMIC_FAULT /usr/bin/php "$R" --authenticated-vessel "$VESSEL" < "$1"; }
assert_db() { local got;got=$(q "$1"); [[ $got == "$2" ]] || { echo "FAIL: expected [$2] got [$got]";exit 1; }; }
parallel() {
    local name=$1 times=$2 fixture=$3 i
    local -a pids=()
    for ((i=0;i<times;i++)); do
        (run "$fixture" > "$OUT/$name.$i.out" 2> "$OUT/$name.$i.err") & pids+=("$!")
    done
    for pid in "${pids[@]}"; do wait "$pid" || { echo "FAIL: concurrent invocation failed in $name"; cat "$OUT/$name".*.err; echo "Logs: $OUT"; exit 1; }; done
    python3 - "$OUT" "$name" "$times" <<'PYT'
import json, pathlib, sys
folder,case,n=pathlib.Path(sys.argv[1]),sys.argv[2],int(sys.argv[3])
statuses=[json.loads((folder/f'{case}.{i}.out').read_text())['status'] for i in range(n)]
assert statuses.count('received') == 1 and statuses.count('already_received') == n-1, statuses
print(f'{case}: received=1, already_received={n-1}')
PYT
}
echo '=== A. Parallel identical manifest: exactly one receipt ==='
parallel header 12 fixtures/header.json
assert_db "SELECT COUNT(*) FROM sync_receipts WHERE vessel_id=UUID_TO_BIN('$VESSEL') AND outbox_seq_no=1" 1
assert_db "SELECT COUNT(*) FROM signalk_catalog_snapshots_v2 WHERE vessel_id=UUID_TO_BIN('$VESSEL')" 1

echo '=== B. Parallel identical chunks ==='
parallel chunk0 12 fixtures/chunk-000.json
parallel chunk1 12 fixtures/chunk-001.json
assert_db "SELECT COUNT(*) FROM signalk_catalog_chunks_v2 WHERE snapshot_id=UUID_TO_BIN('$SID')" 2
assert_db "SELECT COUNT(*) FROM sync_receipts WHERE vessel_id=UUID_TO_BIN('$VESSEL')" 3

echo '=== C. Parallel finalization: exactly one active revision and receipt ==='
parallel finalize 12 fixtures/finalize.json
assert_db "SELECT COUNT(*) FROM signalk_catalog_active_v2 WHERE vessel_id=UUID_TO_BIN('$VESSEL') AND revision=1" 1
assert_db "SELECT COUNT(*) FROM sync_receipts WHERE vessel_id=UUID_TO_BIN('$VESSEL')" 4
assert_db "SELECT COUNT(*) FROM signalk_catalog_snapshots_v2 WHERE vessel_id=UUID_TO_BIN('$VESSEL') AND status='active'" 1

echo '=== D. Simultaneous conflicting messages with same seq_no ==='
(set +e; sudo -n env -u YM_CATALOG_ATOMIC_FAULT /usr/bin/php "$R" --authenticated-vessel "$VESSEL" < fixtures/conflict-header-a.json > "$OUT/conflict-a.out" 2> "$OUT/conflict-a.err"; echo $? > "$OUT/conflict-a.rc") & PA=$!
(set +e; sudo -n env -u YM_CATALOG_ATOMIC_FAULT /usr/bin/php "$R" --authenticated-vessel "$VESSEL" < fixtures/conflict-header-b.json > "$OUT/conflict-b.out" 2> "$OUT/conflict-b.err"; echo $? > "$OUT/conflict-b.rc") & PB=$!
wait "$PA";wait "$PB"
python3 - "$OUT" <<'PYT'
from pathlib import Path
import json,sys
p=Path(sys.argv[1]);rc=[int((p/f'conflict-{x}.rc').read_text()) for x in 'ab']
assert sorted(rc)==[0,1], {"rc":rc,"stderr_a":(p/"conflict-a.err").read_text(),"stderr_b":(p/"conflict-b.err").read_text()}
winner='a' if rc[0]==0 else 'b';loser='b' if winner=='a' else 'a'
assert json.loads((p/f'conflict-{winner}.out').read_text())['status']=='received'
assert 'sync receipt sequence/hash conflict' in (p/f'conflict-{loser}.err').read_text()
print('conflict: exactly one received; conflicting hash rejected')
PYT
assert_db "SELECT COUNT(*) FROM sync_receipts WHERE vessel_id=UUID_TO_BIN('$VESSEL') AND outbox_seq_no=20" 1
assert_db "SELECT COUNT(*) FROM signalk_catalog_snapshots_v2 WHERE vessel_id=UUID_TO_BIN('$VESSEL') AND revision IN (2,3)" 1
python3 - "$OUT" "$DB" "$VESSEL" <<'PYHASH'
from pathlib import Path
import json,subprocess,sys
p,db,v=Path(sys.argv[1]),sys.argv[2],sys.argv[3]
w='a' if int((p/'conflict-a.rc').read_text())==0 else 'b'
expected=json.loads(Path(f'fixtures/conflict-header-{w}.json').read_text())['content_hash']
actual=subprocess.check_output(['sudo','mysql','-N',db,'-e',f"SELECT LOWER(HEX(content_hash)) FROM sync_receipts WHERE vessel_id=UUID_TO_BIN('{v}') AND outbox_seq_no=20"],text=True).strip()
assert expected==actual,(expected,actual)
print(f'conflict: winning receipt hash matches fixture {w}')
PYHASH
assert_db "SELECT COUNT(*) FROM signalk_catalog_snapshots_v2 WHERE vessel_id=UUID_TO_BIN('$VESSEL')" 2

echo '=== E. Two different sequence numbers in parallel ==='
(sudo -n env -u YM_CATALOG_ATOMIC_FAULT /usr/bin/php "$R" --authenticated-vessel "$VESSEL" < fixtures/parallel-header-a.json > "$OUT/distinct-a.out" 2> "$OUT/distinct-a.err") & PA=$!
(sudo -n env -u YM_CATALOG_ATOMIC_FAULT /usr/bin/php "$R" --authenticated-vessel "$VESSEL" < fixtures/parallel-header-b.json > "$OUT/distinct-b.out" 2> "$OUT/distinct-b.err") & PB=$!
wait "$PA";wait "$PB"
python3 - "$OUT" <<'PYT'
from pathlib import Path
import json,sys
p=Path(sys.argv[1]);got=[json.loads((p/f'distinct-{x}.out').read_text())['status'] for x in 'ab']
assert got==['received','received'],got
print('distinct seq_no: both received')
PYT
assert_db "SELECT COUNT(*) FROM sync_receipts WHERE vessel_id=UUID_TO_BIN('$VESSEL')" 7
assert_db "SELECT COUNT(*) FROM signalk_catalog_snapshots_v2 WHERE vessel_id=UUID_TO_BIN('$VESSEL')" 4
assert_db "SELECT revision FROM signalk_catalog_active_v2 WHERE vessel_id=UUID_TO_BIN('$VESSEL')" 1
echo '=== CONCURRENT MYSQL TEST PASS ==='
echo "Logs retained: $OUT"