<?php

namespace Tests\Benchmarks;

use App\Models\User;
use Illuminate\Support\Facades\DB;
use Tests\Support\IdentityFixture;

require_once __DIR__.'/ModuleAssociationLoadTest.php';

class ModuleAssociationConcurrencyTest extends ModuleAssociationLoadTest
{
    protected array $jobs = [];
    protected string $directory;
    protected int $serial = 0;

    public function test_synthetic_volume_and_query_plans(): void
    {
        $run = (string) getenv('SGF_ISOLATED_REGRESSION_RUN');
        $this->assertMatchesRegularExpression('/^[a-f0-9]{12}$/D', $run);
        $identity = DB::selectOne("SELECT current_database() AS database,current_user AS role,inet_server_port() AS port,current_setting('cluster_name') AS cluster");
        $this->assertSame('sgf_tempo_test', $identity->database); $this->assertSame('sgf_test_runner', $identity->role);
        $this->assertSame('sgf_modular_regression_'.$run, $identity->cluster); $this->assertSame(55439, (int) $identity->port);
        $tag = bin2hex(random_bytes(6)); $index = 'sgf_qa_read_revision_'.$tag;
        $this->directory = base_path('../private/association-concurrency-'.$tag); mkdir($this->directory, 0700, true);
        $before = $this->baseline();
        $this->assertSame(0, (int) DB::selectOne("SELECT count(*) AS n FROM pg_indexes WHERE schemaname='public' AND tablename='identity_audit' AND indexname <> 'identity_audit_pkey'")->n, 'Unindexed baseline must be explicit');
        $report = ['date' => '2026-11-02', 'run_tag' => $tag, 'identity' => $identity, 'optimized_sha256' => hash_file('sha256', base_path('app/Modules/Identity/Services/ModuleAssociationReview.php')),
            'method' => 'Independent PHP processes and synthetic identities, pre-authenticated in-process HTTP kernels, synchronized starts, closed loop of five reads per worker, three waves; no external HTTP/login/browser capacity claim.',
            'fixture_policy' => 'New synthetic scope committed only in regression DB for subprocess visibility; retained with private manifest. Existing rows preserved, not reset or deleted.', 'phases' => []];
        $f = $this->fixture(3000, true); $scope = $f['identity'];
        $fourth = User::create(['name' => 'Consulta carga sintética 4', 'email' => 'assoc-load-'.$tag.'@example.test', 'password' => bin2hex(random_bytes(24))]);
        DB::table('company_memberships')->insert(['user_id' => $fourth->id, 'tenant_id' => $scope['tenant'], 'company_id' => $scope['company']]);
        $actors = [$scope['manager']->id, $scope['reader']->id, $scope['limited']->id, $fourth->id];
        foreach ($actors as $actor) {
            foreach (['company.view', 'access.manage'] as $permission) IdentityFixture::grant($actor, $scope['tenant'], $scope['company'], $permission);
            IdentityFixture::grant($actor, $scope['tenant'], $scope['company'], 'unit.view', $scope['unit']);
            foreach (['clients', 'services', 'service_drafts', 'billing_drafts'] as $area) IdentityFixture::grant($actor, $scope['tenant'], $scope['company'], "$area.view", str_ends_with($area, '_drafts') ? $scope['unit'] : null);
        }
        IdentityFixture::grant($actors[0], $scope['tenant'], $scope['company'], 'clients.edit');
        for ($i = 0; $i < 25; $i++) {
            $lines = []; foreach (array_slice($f['ids']['services'], $i * 100, 100) as $id) $lines[] = ['service_id' => $id, 'quantity' => '1', 'unit_price' => '10', 'description' => str_repeat('x', 1000)];
            $data = ['currency' => 'MZN', 'tax_status' => 'pending', 'accounting_status' => 'not_processed', 'lines' => $lines];
            DB::table('a_service_drafts')->where('id', $f['ids']['service_drafts'][$i])->update(['version' => 2, 'data' => json_encode($data)]);
            $data['source_version'] = 2; DB::table('a_billing_drafts')->where('id', $f['ids']['billing_drafts'][$i])->update(['version' => 2, 'data' => json_encode($data)]);
        }
        foreach (['a_service_drafts', 'a_billing_drafts', 'identity_audit'] as $table) DB::statement('ANALYZE '.$table);
        $this->actingAs($scope['manager'])->withSession(['auth_version' => $scope['manager']->auth_version]);
        $context = $this->postJson('/api/context', ['tenant_id' => $scope['tenant'], 'company_id' => $scope['company'], 'unit_id' => $scope['unit'], 'module_id' => null])->assertOk()->json('context.token');
        $this->withHeader('X-SGF-Context', $context);
        $workloads = [];
        foreach (['clients', 'billing_drafts'] as $area) {
            $url = '/api/module-records?area='.$area.'&per_page=25'; $r = $this->getJson($url)->assertOk()->assertJsonPath('pagination.total', 3000);
            $workloads[$area] = ['url' => $url, 'total' => 3000, 'response_sha256' => hash('sha256', $r->getContent())];
        }
        // Separate processes cannot see uncommitted rows. Only this new QA scope is committed.
        DB::commit();
        file_put_contents($this->directory.'/manifest.json', json_encode(['synthetic_only' => true, 'tenant' => $scope['tenant'], 'company' => $scope['company'], 'unit' => $scope['unit'], 'actors' => $actors, 'workloads' => $workloads, 'index_name' => $index], JSON_PRETTY_PRINT | JSON_THROW_ON_ERROR));
        $created = false;
        try {
            foreach (['none', 'revision'] as $mode) {
                if ($mode === 'revision') {
                    $this->assertNull(DB::selectOne('SELECT to_regclass(?) AS value', [$index])->value);
                    DB::statement("CREATE INDEX $index ON identity_audit (tenant_id,company_id,(details->>'area'),(details->>'record_id'),id DESC) WHERE action IN ('module.record.associated','module.record.disassociated','module.record.reviewed')"); $created = true;
                    DB::statement('ANALYZE identity_audit');
                }
                $phase = ['index_mode' => $mode, 'waves' => []];
                foreach ($workloads as $label => $workload) foreach ([1, 2, 4] as $count) for ($wave = 0; $wave < 3; $wave++) {
                    $phase['waves'][] = ['workload' => $label, 'configured_workers' => $count, ...$this->wave($scope, array_slice($actors, 0, $count), $workload, 5, $tag)];
                    fwrite(STDOUT, "\nCONCURRENT $mode $label workers=$count wave=".($wave + 1)." complete\n");
                }
                $report['phases'][] = $phase;
            }
            $candidate = array_values(array_filter($f['ids']['clients'], fn ($id) => $id % 3 === 0))[0];
            $input = ['area' => 'clients', 'id' => $candidate, 'version' => 1, 'operation' => 'associate', 'module_id' => 'commercial', 'reason' => 'Synthetic concurrency rollback only'];
            $first = $this->postJson('/api/module-records/preview', $input)->assertOk()->json('review_token');
            $second = $this->postJson('/api/module-records/preview', $input)->assertOk()->json('review_token');
            $events = DB::table('identity_audit')->where('tenant_id', $scope['tenant'])->where('company_id', $scope['company'])->where('action', 'like', 'module.record.%')->count();
            DB::beginTransaction();
            try {
                $this->postJson('/api/module-records', ['review_token' => $first])->assertOk();
                $report['mixed_read_wave'] = $this->wave($scope, $actors, $workloads['clients'], 5, $tag);
                $gate = $this->directory.'/confirm-go';
                $job = $this->launch(['mode' => 'confirm', 'actor' => $actors[0], 'tenant' => $scope['tenant'], 'company' => $scope['company'], 'unit' => $scope['unit'], 'context' => $context,
                    'review_token' => $second, 'go' => $gate, 'application_name' => 'sgf_assoc_confirm_'.$tag]);
                $this->ready([$job]); file_put_contents($gate, 'go'); $observed = false; $deadline = microtime(true) + 3;
                do { DB::select('SELECT pg_stat_clear_snapshot()'); $n = DB::selectOne("SELECT count(*) AS n FROM pg_stat_activity WHERE application_name=? AND wait_event_type='Lock'", ['sgf_assoc_confirm_'.$tag])->n; if ((int) $n > 0) { $observed = true; break; } usleep(25000); } while (microtime(true) < $deadline);
                $result = $this->collect($job); $this->assertTrue($observed, 'Actual PostgreSQL lock wait must be observed');
                $this->assertSame('55P03', $result['confirmation']['sqlstate'] ?? null);
                $report['competing_confirmation'] = ['actual_lock_wait_observed' => $observed, ...$result['confirmation']];
            } finally { DB::rollBack(); }
            $this->assertSame($events, DB::table('identity_audit')->where('tenant_id', $scope['tenant'])->where('company_id', $scope['company'])->where('action', 'like', 'module.record.%')->count());
            $report['synthetic_confirmation_rolled_back'] = true;
        } finally {
            foreach ($this->jobs as $job) if (is_resource($job['process'])) { proc_terminate($job['process']); proc_close($job['process']); }
            while (DB::transactionLevel() > 0) DB::rollBack();
            if ($created) DB::statement('DROP INDEX '.$index);
            $this->assertNull(DB::selectOne('SELECT to_regclass(?) AS value', [$index])->value);
            $report['test_index_removed'] = true;
            $report['previous_rows_unchanged'] = $this->baseline($before) === $before;
            $this->assertTrue($report['previous_rows_unchanged']);
            $output = getenv('SGF_ASSOCIATION_LOAD_OUTPUT');
            if (!$output || realpath(dirname($output)) !== realpath(base_path('..'))) throw new \RuntimeException('Unexpected report directory');
            file_put_contents($output, json_encode($report, JSON_PRETTY_PRINT | JSON_THROW_ON_ERROR).PHP_EOL);
            // Preserve the transaction-based TestCase lifecycle without undoing committed QA setup.
            DB::beginTransaction();
        }
    }

    private function wave(array $scope, array $actors, array $workload, int $requests, string $tag): array
    {
        $gate = $this->directory.'/go-'.(++$this->serial); $jobs = []; $startup = microtime(true);
        foreach ($actors as $i => $actor) $jobs[] = $this->launch(['mode' => 'read', 'actor' => $actor, 'tenant' => $scope['tenant'], 'company' => $scope['company'], 'unit' => $scope['unit'],
            ...$workload, 'requests' => $requests, 'go' => $gate, 'application_name' => 'sgf_assoc_read_'.$tag.'_'.$i]);
        $this->ready($jobs); $ready = microtime(true); file_put_contents($gate, 'go');
        $samples = []; $finished = $ready;
        foreach ($jobs as $job) { $result = $this->collect($job); $samples = [...$samples, ...$result['requests']]; $finished = max($finished, $result['finished_at']); }
        $ok = array_values(array_filter($samples, fn ($s) => $s['status'] === 'ok')); $events = [];
        foreach ($ok as $s) { $events[] = [$s['started_at'], 1]; $events[] = [$s['finished_at'], -1]; }
        usort($events, fn ($a, $b) => ($a[0] <=> $b[0]) ?: ($a[1] <=> $b[1])); $active = 0; $max = 0;
        foreach ($events as $event) { $active += $event[1]; $max = max($max, $active); }
        $this->assertLessThanOrEqual(count($actors), $max); $this->assertGreaterThanOrEqual(1, $max);
        return ['startup_to_ready_ms' => round(($ready - $startup) * 1000, 3), 'measurement_window_ms' => round(($finished - $ready) * 1000, 3), 'observed_max_inflight' => $max,
            'complete_requests' => count($ok), 'limited_requests' => count($samples) - count($ok), 'samples' => $samples];
    }

    protected function launch(array $input): array
    {
        $stem = $this->directory.'/worker-'.(++$this->serial);
        $input += ['ready' => $stem.'.ready', 'result' => $stem.'.json'];
        file_put_contents($stem.'.input.json', json_encode($input, JSON_THROW_ON_ERROR));
        $env = getenv(); $env['SGF_ASSOCIATION_WORKER_INPUT'] = $stem.'.input.json';
        $pipes = []; $process = proc_open([PHP_BINARY, 'vendor/phpunit/phpunit/phpunit', 'scripts/benchmarks/ModuleAssociationReadWorkerTest.php', '--no-progress'],
            [0 => ['pipe', 'r'], 1 => ['file', $stem.'.stdout', 'w'], 2 => ['file', $stem.'.stderr', 'w']], $pipes, base_path(), $env, ['bypass_shell' => true]);
        $this->assertIsResource($process); fclose($pipes[0]);
        $job = ['process' => $process, 'input' => $input, 'stdout' => $stem.'.stdout', 'stderr' => $stem.'.stderr']; $this->jobs[] = $job; return $job;
    }

    protected function ready(array $jobs): void
    {
        $deadline = microtime(true) + 30;
        do {
            $all = true;
            foreach ($jobs as $job) if (!is_file($job['input']['ready'])) { $all = false; if (!proc_get_status($job['process'])['running']) throw new \RuntimeException('Worker failed before ready: '.file_get_contents($job['stdout']).file_get_contents($job['stderr'])); }
            if ($all) return; usleep(10000);
        } while (microtime(true) < $deadline);
        throw new \RuntimeException('Workers did not reach barrier');
    }

    protected function collect(array $job): array
    {
        $deadline = microtime(true) + 60;
        do { $status = proc_get_status($job['process']); if (!$status['running']) break; usleep(10000); } while (microtime(true) < $deadline);
        if ($status['running']) throw new \RuntimeException('Worker deadline reached');
        $exit = proc_close($job['process']); if ($exit === -1) $exit = $status['exitcode'];
        $this->assertSame(0, $exit, file_get_contents($job['stdout']).file_get_contents($job['stderr']));
        return json_decode(file_get_contents($job['input']['result']), true, 512, JSON_THROW_ON_ERROR);
    }

    protected function baseline(?array $old = null): array
    {
        $out = [];
        foreach (['a_clients', 'a_services', 'a_service_drafts', 'a_billing_drafts', 'identity_audit', 'users', 'permission_grants', 'company_memberships'] as $table) {
            $key = $table === 'company_memberships' ? 'user_id' : 'id';
            $q = DB::table($table);
            // IDs are captured integer primary keys, not user input. Avoid the
            // PostgreSQL bind-parameter ceiling in accumulated QA fixtures.
            if ($old !== null) $q->whereIntegerInRaw($key, $old[$table]['ids']);
            $q->orderBy($key);
            if ($table === 'company_memberships') $q->orderBy('tenant_id')->orderBy('company_id');
            $rows = $q->get();
            $out[$table] = ['ids' => $old[$table]['ids'] ?? $rows->pluck($key)->unique()->values()->all(), 'hash' => hash('sha256', $rows->toJson())];
        }
        $out['indexes'] = json_encode(DB::select("SELECT indexname,indexdef FROM pg_indexes WHERE schemaname='public' ORDER BY indexname"), JSON_THROW_ON_ERROR);
        return $out;
    }
}