<?php

namespace App\Services;

use App\Models\Account;
use App\Models\Bank;
use App\Models\TransactionType;
use App\Services\Stats\LocationFilterResolver;
use App\Services\Stats\SlidingWindow;
use App\Services\Stats\StatsFilter;
use Carbon\Carbon;
use Carbon\CarbonPeriod;
use Illuminate\Support\Facades\Cache;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;

/**
 * Platform-wide statistics over a filtered set of profiles.
 *
 * Every metric is a GROUP BY aggregate: transaction rows are never loaded into
 * PHP. The per-profile report (App\Http\Livewire\SingleReport) does load rows
 * and issues one balance query per month per account; at platform scale that
 * exhausts memory and floods the DB, so none of that is reused here.
 */
class StatsService
{
    /**
     * Above this many account ids, WHERE IN lists stop being a good idea and the
     * resolved set should be materialized into a temporary table instead.
     */
    public const ID_MATERIALIZE_LIMIT = 50000;

    private const CACHE_TTL = 900; // 15 minutes
    /**
     * Bump this whenever the aggregation logic changes — filter-hashed keys
     * cannot be selectively invalidated, so the version prefix retires them.
     */
    private const CACHE_VERSION = 'v43';

    /**
     * Entries the system makes itself, which therefore prove no activity by the
     * profile: currency creation/removal, migration, rounding correction and
     * reversal. Real trade = Work, Gift, Donation.
     */
    private const SYSTEM_TRANSACTION_TYPES = [4, 5, 6, 7, 8];

    /**
     * Last moment a transaction could have come from Cyclos.
     *
     * MigrateCyclosCommand copies `transfers.type_id` verbatim (see the INSERT
     * around line 925), so migrated rows carry CYCLOS transfer-type ids, not
     * the local ones. The two sets overlap and disagree: Cyclos 1 is "Debit to
     * community" where local 1 is "Work". Everything from this instant onwards
     * was written by this application and uses the local ids.
     *
     * Verified on the data: the last id > 8 is 2026-03-26 20:55:34 and the
     * first native row is the next afternoon. The Cyclos snapshot database is
     * even named timebank_2026_03_27.
     */
    public const CYCLOS_CUTOVER = '2026-03-26 20:55:34';

    /**
     * Cyclos transfer types excluded from the graph.
     *
     * Names from timebank_2026_03_27.transfer_types. Deliberately SHORT: the
     * Cyclos equivalents of currency creation, currency removal and migration
     * stay IN, because the graph is meant to show where hours enter and leave
     * the network, not only member-to-member trade. Their local counterparts
     * (4, 5, 6) are handled by the post-cutover branch, which is why the two
     * lists are not mirror images.
     *
     * Mapped with the user, 2026-09-26:
     *    1  Debit to community  -> currency creation   (kept)
     *    4  Community to debit  -> currency removal    (kept)
     *   15  Community to local bank    -> migration    (kept)
     *   33  System group to local bank -> migration    (kept)
     *   52  Currency creation for an User             (kept)
     *   53  Currency creation for a Project           (kept)
     *
     * Only self-payments are dropped: they move hours between one member's own
     * accounts, so counting them would let a profile inflate its own volume
     * and would draw a self-loop the layout cannot render meaningfully.
     *
     *   42  Self payment: User to Gift Account
     *   50  Self payment: Work to Project Account
     *   51  Self payment: ~Gift to Project Account
     */
    private const SYSTEM_CYCLOS_TYPES = [42, 50, 51];

    /**
     * Local types the GRAPH excludes. Only rounding corrections and reversals:
     * bookkeeping noise that describes no movement between two profiles.
     *
     * Deliberately NOT SYSTEM_TRANSACTION_TYPES. Currency creation (4),
     * removal (5) and migration (6) stay in, because the graph is there to
     * show where hours enter and leave the network. Excluding them left the
     * banks' counterparties looking like unexplained sources.
     *
     * The inactivity filter keeps using SYSTEM_TRANSACTION_TYPES: "has this
     * profile traded lately" is a different question, and being handed newly
     * created currency is not trading.
     */
    private const GRAPH_EXCLUDED_LOCAL_TYPES = [7, 8];

    /**
     * Canonical transaction type as SQL, folding the Cyclos ids onto the local
     * ones so an edge can be labelled with a name people recognise.
     *
     * Names from timebank_2026_03_27.transfer_types, mapped with the user:
     *   Work     45 "1. Work paid from your Project Account", 48, 13, 49, 37
     *   Gift     41 "3. Gift paid from your Work Account", 47, 40
     *   Donation 46 "1. Donation paid from your Project Account", 43, 44
     *   Creation 52, 53, 1 ("Debit to community")
     *   Removal  4 ("Community to debit")
     *   Migration 15, 33 (inter-bank), 14, 18
     *
     * @param string $alias Table alias carrying transaction_type_id/created_at
     */
    private function canonicalTypeSql(string $alias = 't'): string
    {
        $cut = self::CYCLOS_CUTOVER;

        return "(CASE WHEN {$alias}.created_at <= '{$cut}' THEN CASE
                    WHEN {$alias}.transaction_type_id IN (45,48,13,49,37) THEN 1
                    WHEN {$alias}.transaction_type_id IN (41,47,40)       THEN 2
                    WHEN {$alias}.transaction_type_id IN (46,43,44)       THEN 3
                    WHEN {$alias}.transaction_type_id IN (52,53,1)        THEN 4
                    WHEN {$alias}.transaction_type_id = 4                 THEN 5
                    WHEN {$alias}.transaction_type_id IN (15,33,14,18)    THEN 6
                    -- Written by MigrateCyclosCommand itself (it sets 7
                    -- explicitly), not copied from Cyclos: that database has
                    -- zero type-7 transfers, and all 656 of ours say
                    -- \"rounding discrepancy\" at 0.066h. So local 7, not
                    -- Cyclos \"Voucher destruction\".
                    WHEN {$alias}.transaction_type_id IN (7, 8)           THEN {$alias}.transaction_type_id
                    -- Self-payments and mobile transfers: no local equivalent.
                    ELSE 0 END
                ELSE {$alias}.transaction_type_id END)";
    }

    /**
     * SQL predicate selecting the transactions the graph draws.
     *
     * Filters on the CANONICAL type, so one list covers both eras instead of
     * two that had to be kept in step. Excluded: rounding corrections and
     * reversals (bookkeeping noise describing no movement between profiles)
     * and type 0, which is what canonicalTypeSql() returns for self-payments
     * and mobile transfers that have no local equivalent.
     *
     * Currency creation, removal and migration stay IN: the graph is meant to
     * show where hours enter and leave the network, not only member-to-member
     * trade.
     *
     * @param string $alias Table alias carrying transaction_type_id/created_at
     */
    private function graphTypesSql(string $alias = 't'): string
    {
        return $this->canonicalTypeSql($alias) . ' NOT IN (0, 7, 8)';
    }

    /**
     * SQL predicate for REAL TRADE, per era. Used by the inactivity filter,
     * where a system booking must not count as activity.
     *
     * @param string $alias Table alias carrying transaction_type_id/created_at
     */
    private function realTradeSql(string $alias = 't'): string
    {
        $local  = implode(',', self::SYSTEM_TRANSACTION_TYPES);
        // Pre-cutover equivalents of currency creation, removal and migration,
        // plus self-payments. See SYSTEM_CYCLOS_TYPES for the names.
        $cyclos = implode(',', [52, 53, 15, 33, 1, 4, 14, 18, 42, 50, 51]);
        $cut    = self::CYCLOS_CUTOVER;

        return "(CASE WHEN {$alias}.created_at <= '{$cut}'"
             . " THEN {$alias}.transaction_type_id NOT IN ({$cyclos})"
             . " ELSE {$alias}.transaction_type_id NOT IN ({$local}) END)";
    }

    /**
     * Top stop of the degree slider. Above it there is no bound at all: on dev
     * only 3 nodes have more than 50 partners against a maximum of 700, so a
     * linear scale up to 700 would cram 99% of the nodes into the first
     * percent of the slider.
     */
    public const DEGREE_CEILING = 50;

    /** @var array<class-string, array<int>>|null */
    private ?array $profileKeys = null;

    /** @var array<int>|null */
    private ?array $accountIds = null;

    public function __construct(private StatsFilter $filter) {}

    public function all(bool $fresh = false): array
    {
        // Keyed by locale: the payload carries translated strings (transaction
        // type names on every graph edge, chart labels), so one cache shared
        // across locales served Dutch names to an English session.
        $key = 'stats:' . self::CACHE_VERSION . ':' . app()->getLocale()
             . ':' . $this->filter->hash();

        if ($fresh) {
            Cache::forget($key);
        }

        return Cache::remember($key, self::CACHE_TTL, fn() => [
            'economic'     => $this->economic(),
            'online'       => $this->online(),
            'engagement'   => $this->engagement(),
            'distribution' => $this->distribution(),
            'balance'      => $this->balance(),
            'balance_distribution' => $this->balanceDistribution(),
            'profileCount' => collect($this->profileKeys())->map(fn($ids) => count($ids))->sum(),
            'accountCount' => count($this->accountIds()),
        ]);
    }

    // ---------------------------------------------------------------- resolution

    /**
     * Profiles matching the filter, grouped by model class. Resolved once.
     *
     * @return array<class-string, array<int>>
     */
    public function profileKeys(): array
    {
        if ($this->profileKeys !== null) {
            return $this->profileKeys;
        }

        $keys = [];

        foreach (StatsFilter::PROFILE_TYPES as $type) {
            // withoutGlobalScopes() is needed as well as dropping notRemoved():
            // the models soft-delete, so the global scope would filter closed
            // profiles out again before notRemoved() ever ran.
            $query = $this->filter->includeRemoved
                ? $type::query()->withoutGlobalScopes()
                : $type::query()->notRemoved();

            if ($this->filter->registeredFrom) {
                $query->where('created_at', '>=', $this->filter->registeredFrom);
            }
            if ($this->filter->registeredTill) {
                $query->where('created_at', '<=', $this->filter->registeredTill);
            }

            $query = LocationFilterResolver::apply(
                $query,
                $this->filter->countryIds,
                $this->filter->divisionIds,
                $this->filter->cityIds,
                $this->filter->districtIds
            );

            if (!empty($this->filter->counterparties)) {
                $ids = $this->profileIdsTransactingWithCounterparties($type);
                $query->whereIn('id', $ids);
            }

            // Not logged in since: profiles that never logged in count too. A
            // plain `<` would silently drop them.
            if ($this->filter->notLoggedInSince) {
                $table = (new $type)->getTable();

                if (\Schema::hasColumn($table, 'last_login_at')) {
                    $query->where(function ($q) {
                        $q->whereNull('last_login_at')
                          ->orWhere('last_login_at', '<', $this->filter->notLoggedInSince);
                    });
                } else {
                    // Banks have no login: under this filter they drop out
                    // rather than passing through unfiltered.
                    $query->whereRaw('1 = 0');
                }
            }

            $keys[$type] = $query->pluck('id')->map('intval')->all();
        }

        // No transaction since: handled separately, as it needs the
        // transactions table.
        if ($this->filter->noTransactionSince) {
            $keys = $this->withoutRecentTransactions($keys, $this->filter->noTransactionSince);
        }

        return $this->profileKeys = $keys;
    }

    /**
     * Account ids belonging to the resolved profile set.
     *
     * @return array<int>
     */
    /**
     * Keeps only profiles without a transaction since $since.
     *
     * System entries do not count as activity: a migration or year-end closing
     * touches every account and would mark everyone "active". Only Work, Gift
     * and Donation count as real trade.
     *
     * @param array<class-string, array<int>> $keys
     * @return array<class-string, array<int>>
     */
    private function withoutRecentTransactions(array $keys, \Carbon\Carbon $since): array
    {
        $candidates = array_filter($keys, fn($ids) => !empty($ids));

        if (empty($candidates)) {
            return $keys;
        }

        $accounts = Account::notRemoved()
            ->where(function ($q) use ($candidates) {
                foreach ($candidates as $type => $ids) {
                    $q->orWhere(function ($w) use ($type, $ids) {
                        $w->where('accountable_type', $type)->whereIn('accountable_id', $ids);
                    });
                }
            })
            ->get(['id', 'accountable_type', 'accountable_id']);

        if ($accounts->isEmpty()) {
            return $keys;
        }

        $in = $this->inList($accounts->pluck('id')->map('intval')->all());

        $active = DB::table('transactions')
            ->whereRaw("(from_account_id IN ($in) OR to_account_id IN ($in))")
            ->whereRaw($this->realTradeSql('transactions'))
            ->where('created_at', '>=', $since)
            ->get(['from_account_id', 'to_account_id']);

        $activeAccounts = [];
        foreach ($active as $row) {
            $activeAccounts[$row->from_account_id] = true;
            $activeAccounts[$row->to_account_id] = true;
        }

        // A profile is active as soon as one of its accounts is.
        $activeProfiles = [];
        foreach ($accounts as $account) {
            if (isset($activeAccounts[$account->id])) {
                $activeProfiles[$account->accountable_type . ':' . $account->accountable_id] = true;
            }
        }

        foreach ($keys as $type => $ids) {
            $keys[$type] = array_values(array_filter(
                $ids,
                fn($id) => !isset($activeProfiles[$type . ':' . $id])
            ));
        }

        return $keys;
    }

    public function accountIds(): array
    {
        if ($this->accountIds !== null) {
            return $this->accountIds;
        }

        $keys = array_filter($this->profileKeys(), fn($ids) => !empty($ids));

        if (empty($keys)) {
            return $this->accountIds = [];
        }

        $ids = $this->accountScope()
            ->where(function ($q) use ($keys) {
                foreach ($keys as $type => $profileIds) {
                    $q->orWhere(function ($w) use ($type, $profileIds) {
                        $w->where('accountable_type', $type)->whereIn('accountable_id', $profileIds);
                    });
                }
            })
            ->pluck('id')
            ->map('intval')
            ->all();

        if (count($ids) > self::ID_MATERIALIZE_LIMIT) {
            Log::warning('StatsService: resolved account set exceeds the WHERE IN ceiling; consider a temporary table', [
                'count' => count($ids),
                'limit' => self::ID_MATERIALIZE_LIMIT,
            ]);
        }

        return $this->accountIds = $ids;
    }

    /**
     * Profiles of $type that transacted with any of the selected counterparties
     * during the period. Resolved entirely in SQL.
     */
    private function profileIdsTransactingWithCounterparties(string $type): \Illuminate\Support\Collection
    {
        $counterpartyAccountIds = Account::notRemoved()
            ->where(function ($q) {
                foreach ($this->filter->counterparties as $party) {
                    $q->orWhere(function ($w) use ($party) {
                        $w->where('accountable_type', $party['type'])
                          ->where('accountable_id', (int) $party['id']);
                    });
                }
            })
            ->pluck('id');

        if ($counterpartyAccountIds->isEmpty()) {
            return collect();
        }

        $in = $this->inList($counterpartyAccountIds->map('intval')->all());

        $touchedAccountIds = $this->applyPeriod(DB::table('transactions'))
            ->where(function ($q) use ($counterpartyAccountIds) {
                $q->whereIn('from_account_id', $counterpartyAccountIds)
                  ->orWhereIn('to_account_id', $counterpartyAccountIds);
            })
            ->selectRaw("DISTINCT IF(from_account_id IN ($in), to_account_id, from_account_id) AS aid")
            ->pluck('aid');

        $ids = $touchedAccountIds->isEmpty()
            ? collect()
            : Account::whereIn('id', $touchedAccountIds)
                ->where('accountable_type', $type)
                ->distinct()
                ->pluck('accountable_id');

        // The selected profiles belong in their own selection: the query above
        // only returns the *other* side of each transaction, so without this the
        // person you filtered on would be missing from their own statistics.
        $selected = collect($this->filter->counterparties)
            ->where('type', $type)
            ->pluck('id')
            ->map('intval');

        return $ids->merge($selected)->unique()->values();
    }

    /**
     * Account ids of the explicitly selected counterparty profiles.
     *
     * @return array<int>
     */
    private function counterpartyAccountIds(): array
    {
        if (empty($this->filter->counterparties)) {
            return [];
        }

        return Account::notRemoved()
            ->where(function ($q) {
                foreach ($this->filter->counterparties as $party) {
                    $q->orWhere(function ($w) use ($party) {
                        $w->where('accountable_type', $party['type'])
                          ->where('accountable_id', (int) $party['id']);
                    });
                }
            })
            ->pluck('id')
            ->map('intval')
            ->all();
    }

    /**
     * Removed / inactive state per profile key.
     *
     * Two distinct states, as the UI wording has it: a profile CLOSED by its
     * owner carries deleted_at, while one MARKED INACTIVE carries inactive_at.
     * A profile can be neither, either, or both.
     *
     * @param array<string> $keys 'App\Models\User:161'
     * @return array<string, array{removed:bool, inactive:bool}>
     */
    private function profileStates(array $keys): array
    {
        $byType = [];
        foreach ($keys as $key) {
            [$type, $id] = explode(':', $key, 2);
            $byType[$type][] = (int) $id;
        }

        $out = [];

        foreach ($byType as $type => $ids) {
            if (!class_exists($type)) {
                continue;
            }

            $rows = $type::withoutGlobalScopes()
                ->whereIn('id', $ids)
                ->get(['id', 'deleted_at', 'inactive_at']);

            foreach ($rows as $row) {
                $removed = $row->deleted_at !== null && $row->deleted_at <= now();

                $out[$type . ':' . $row->id] = [
                    'removed'  => $removed,
                    // Mutually exclusive: a closed profile keeps whatever
                    // inactive_at it had, so 144 of 147 "inactive" nodes were
                    // really removed ones wearing the wrong marker. Removal is
                    // the final state; only a profile that still exists can be
                    // meaningfully flagged inactive.
                    'inactive' => !$removed
                        && $row->inactive_at !== null
                        && $row->inactive_at <= now(),
                ];
            }
        }

        return $out;
    }

    /**
     * Base account query, honouring the "include removed" toggle.
     *
     * Account::notRemoved() is what keeps currency removals out of the graph:
     * every one is a balance transfer made while closing a profile, so the
     * account is deleted in the same operation and the edge never appears.
     */
    private function accountScope()
    {
        return $this->filter->includeRemoved
            ? Account::withoutGlobalScopes()
            : Account::notRemoved();
    }

    /**
     * Restrict a transaction query to the filtered set.
     *
     * Normally "touches any account in the set" is what we want. But when the
     * user filtered by transactions with specific profiles, they asked about
     * dealings *with those profiles* — counting every unrelated transaction of
     * everyone in the set would let one busy counterparty (an organisation with
     * thousands of its own transactions) swamp the result.
     */
    private function scopeToSet($query, string $in)
    {
        $query->whereRaw("(from_account_id IN ($in) OR to_account_id IN ($in))");

        $counterpartyIds = $this->counterpartyAccountIds();

        if (!empty($counterpartyIds)) {
            $cin = $this->inList($counterpartyIds);
            $query->whereRaw("(from_account_id IN ($cin) OR to_account_id IN ($cin))");
        }

        return $query;
    }

    // ------------------------------------------------------------------ metrics

    /**
     * Volume, counts and per-type breakdown.
     *
     * Volume counts each transaction row once. It deliberately does NOT use the
     * `THEN 0` internal-transfer trick from AccountInfoTrait::getAccountsInfo():
     * that is correct for a *balance* (an internal transfer nets out) but wrong
     * for *volume*, where an internal transfer is real activity. The internal
     * share is reported separately instead of being silently dropped.
     */
    public function economic(): array
    {
        $in = $this->accountIn();

        if ($in === null) {
            return [
                'total_volume' => 0, 'internal_volume' => 0, 'external_volume' => 0,
                'split_is_meaningful' => false,
                'transaction_count' => 0, 'by_type' => collect(), 'timeline' => [],
            ];
        }

        $rows = $this->scopeToSet($this->applyPeriod(DB::table('transactions')), $in)
            ->selectRaw('transaction_type_id')
            ->selectRaw('COUNT(*) AS cnt')
            ->selectRaw('SUM(amount) AS volume')
            ->selectRaw("SUM(CASE WHEN from_account_id IN ($in) AND to_account_id IN ($in) THEN amount ELSE 0 END) AS internal_volume")
            ->groupBy('transaction_type_id')
            ->get();

        $typeNames = $this->transactionTypeNames();

        // Legacy Cyclos-era transactions carry transaction_type_ids with no row in
        // transaction_types. Consolidate them into a single "Unknown" bucket rather
        // than one row per orphan id — matching SingleReport's existing behaviour.
        $byType = $rows
            ->groupBy(fn($r) => $typeNames[$r->transaction_type_id] ?? __('Unknown'))
            ->map(fn($group, $name) => [
                'type_name' => $name,
                'count'     => (int) $group->sum('cnt'),
                'volume'    => (int) $group->sum('volume'),
            ])
            ->sortByDesc('volume')
            ->values();

        $total    = (int) $rows->sum('volume');
        $internal = (int) $rows->sum('internal_volume');

        return [
            'total_volume'      => $total,
            'internal_volume'   => $internal,
            'external_volume'   => $total - $internal,
            // The split only says something when the set can actually have an
            // outside. With a counterparty filter the set is "the selected
            // profiles plus everyone they traded with", so every counted
            // transaction has both ends inside it by construction and external
            // is always 0 — reporting that would imply a finding that isn't one.
            // Unfiltered, the whole platform is the set, so it's zero for the
            // same structural reason.
            'split_is_meaningful' => empty($this->filter->counterparties)
                && ($this->filter->hasLocationFilter() || $this->filter->hasRegistrationFilter()),
            'transaction_count' => (int) $rows->sum('cnt'),
            'by_type'           => $byType,
            'timeline'          => $this->economicTimeline($in),
        ];
    }

    /**
     * Volume and count per month, split per transaction type — one query.
     *
     * Each row carries 'volume'/'count' (all types combined) plus one key per
     * type name, so the chart can group by type while the totals stay intact.
     * Type names are consolidated exactly as the table does, so the legacy
     * Cyclos ids with no transaction_types row share a single "Unknown" series.
     */
    private function economicTimeline(string $in): array
    {
        $rows = $this->scopeToSet($this->applyPeriod(DB::table('transactions')), $in)
            ->selectRaw("DATE_FORMAT(created_at, '%Y-%m') AS ym")
            ->selectRaw('transaction_type_id')
            ->selectRaw('COUNT(*) AS cnt')
            ->selectRaw('SUM(amount) AS volume')
            ->groupBy('ym', 'transaction_type_id')
            ->orderBy('ym')
            ->get();

        if ($rows->isEmpty()) {
            return [];
        }

        $typeNames = $this->transactionTypeNames();
        $unknown   = __('Unknown');

        $byMonth = [];
        foreach ($rows as $row) {
            $name = $typeNames[$row->transaction_type_id] ?? $unknown;
            $byMonth[$row->ym][$name] = ($byMonth[$row->ym][$name] ?? 0) + (int) $row->volume;
            $byMonth[$row->ym]['__count'] = ($byMonth[$row->ym]['__count'] ?? 0) + (int) $row->cnt;
        }

        // Stack order is semantic, not by volume: member exchange at the
        // bottom, then administrative corrections, then currency operations.
        // A volume-sorted stack reshuffles itself whenever the filter changes,
        // which makes two charts impossible to compare by eye.
        $present = [];
        foreach ($byMonth as $values) {
            foreach ($values as $name => $value) {
                if ($name !== '__count') {
                    $present[$name] = true;
                }
            }
        }

        $seriesNames = $this->orderedTypeNames(array_keys($present));

        $empty = array_fill_keys($seriesNames, 0) + ['count' => 0, 'volume' => 0];
        $keyed = collect($byMonth)->map(fn($v, $ym) => (object) ['ym' => $ym])->sortKeys();

        return $this->zeroFillMonths(
            $keyed,
            function ($row) use ($byMonth, $seriesNames) {
                $values = [];
                $volumeMinutes = 0;

                foreach ($seriesNames as $name) {
                    $minutes = $byMonth[$row->ym][$name] ?? 0;
                    $volumeMinutes += $minutes;
                    // Series are charted in hours, matching the currency axis.
                    $values[$name] = $minutes / 60;
                }

                $values['count']  = $byMonth[$row->ym]['__count'] ?? 0;
                $values['volume'] = $volumeMinutes; // kept in minutes, as before
                return $values;
            },
            $empty
        );
    }

    /** Registrations, activity and logins across the resolved profile set. */
    public function online(): array
    {
        $byType = [];
        // Per-type monthly buckets, so the chart can plot User/Organization/Bank
        // separately while the 'count' key keeps the combined total.
        $registrationsByType = [];
        // Same per-type split for inactivity marks, charted cumulatively below.
        $inactiveByType = [];
        // The profiles behind those marks, so their accounts can be resolved
        // for the balance series charted alongside the count.
        $inactiveProfileKeys = [];

        foreach ($this->profileKeys() as $type => $ids) {
            if (empty($ids)) {
                $byType[class_basename($type)] = ['total' => 0, 'active' => 0, 'inactive' => 0, 'recent_logins' => 0];
                continue;
            }

            $table = (new $type)->getTable();
            $hasLogin = \Schema::hasColumn($table, 'last_login_at');

            $query = DB::table($table)->whereIn('id', $ids)
                ->selectRaw('COUNT(*) AS total')
                ->selectRaw('SUM(CASE WHEN inactive_at IS NULL OR inactive_at > NOW() THEN 1 ELSE 0 END) AS active');

            if ($hasLogin) {
                $query->selectRaw('SUM(CASE WHEN last_login_at >= ? THEN 1 ELSE 0 END) AS recent_logins', [now()->subDays(30)]);
            }

            $row = $query->first();

            $byType[class_basename($type)] = [
                'total'         => (int) $row->total,
                'active'        => (int) $row->active,
                'inactive'      => (int) $row->total - (int) $row->active,
                'recent_logins' => $hasLogin ? (int) ($row->recent_logins ?? 0) : null,
            ];

            // The month each profile was marked inactive. A future inactive_at
            // is a scheduled mark that has not taken effect, so it is
            // excluded here exactly as it is from the 'active' count above.
            //
            // Uses the standard $ids, which profileKeys() has already run
            // through notRemoved() — that is the single guard here:
            // a profile that was deleted is removed, not inactive, and its
            // currency has been withdrawn — counting it here would report
            // balances that no longer exist. The two states are distinct even
            // though deletion happens to set both columns.
            //
            // Banks are included, unlike the registrations chart below — a bank
            // being taken out of service is a real event on this timeline,
            // where a bank created at setup is only noise on that one.
            $markedInactive = empty($ids) ? collect() : DB::table($table)
                ->whereIn('id', $ids)
                ->whereNotNull('inactive_at')
                ->where('inactive_at', '<=', now())
                ->get(['id', 'inactive_at']);

            $inactiveByType[class_basename($type)] = $markedInactive
                // The raw query builder returns a string here, but normalise
                // rather than rely on the driver's type for the month key.
                ->groupBy(fn($row) => \Carbon\Carbon::parse($row->inactive_at)->format('Y-m'))
                ->map(fn($group) => $group->count())
                ->all();

            // Keep the ids so their accounts can be resolved for the balance
            // series below — one lookup per type rather than a second pass.
            if ($markedInactive->isNotEmpty()) {
                $inactiveProfileKeys[$type] = $markedInactive->pluck('id')->map('intval')->all();
            }

            // Banks stay in the table above but out of the registrations chart:
            // a handful of banks created at setup would sit as lone spikes on a
            // timeline that is otherwise about members joining.
            if ($type === Bank::class) {
                continue;
            }

            $monthly = DB::table($table)->whereIn('id', $ids)
                ->selectRaw("DATE_FORMAT(created_at, '%Y-%m') AS ym, COUNT(*) AS cnt")
                ->groupBy('ym')
                ->get();

            $registrationsByType[class_basename($type)] = $monthly->pluck('cnt', 'ym')
                ->map(fn($c) => (int) $c)
                ->all();
        }

        return [
            'by_type'       => $byType,
            'total'         => collect($byType)->sum('total'),
            'active'        => collect($byType)->sum('active'),
            'inactive'      => collect($byType)->sum('inactive'),
            'registrations' => $this->buildRegistrationsTimeline($registrationsByType),
            // Present-day figure for the summary tile: how many profiles in
            // the selection stand incomplete now, regardless of period.
            'incomplete'    => $this->incompleteNowCount(),
            'inactivations' => $this->buildInactiveTimeline(
                $inactiveByType,
                $this->inactiveBalanceByMonth($inactiveProfileKeys),
                $this->incompleteByMonth()
            ),
        ];
    }

    /**
     * Spell out when a profile is marked inactive, from the live config.
     *
     * The rule is platform-specific: days_not_logged_in is 350 on the default
     * platform and twenty years here, and re-activation on login is a toggle.
     * Hardcoding "after two years" into a view would quietly go wrong the
     * moment a white-label instance sets a different threshold, so the text is
     * built from the same config values MarkInactiveProfiles reads.
     *
     * Static because the screen and the PDF both need it, and neither the
     * controller nor the component should hold a second copy of the wording.
     */
    /**
     * One short note per charted series, in the order the lines are drawn.
     *
     * Returned as separate strings rather than one block: each paragraph maps
     * to a line on the chart, so a reader can find the one they are looking at
     * instead of parsing a wall of text.
     *
     * @return array<string>
     */
    public static function chartLegendNotes(): array
    {
        $days = (int) timebank_config('profile_inactive.days_not_logged_in');

        // Expressed in whatever unit reads naturally at this platform's scale —
        // "7320 days" is technically right and useless to a reader.
        if ($days >= 365) {
            // Whole years: thresholds are set as round periods (366 * 20 here),
            // and "20,1 years" reads as a precision the setting does not have.
            $years  = (int) round($days / 365.25);
            $period = trans_choice('{1}:count year|[2,*]:count years', $years, ['count' => $years]);
        } elseif ($days >= 60) {
            // Below a year but not near it — 350 days is not "12 months".
            $months = (int) floor($days / 30.44);
            $period = trans_choice('{1}:count month|[2,*]:count months', $months, ['count' => $months]);
        } else {
            $period = trans_choice('{1}:count day|[2,*]:count days', $days, ['count' => $days]);
        }

        $inactive = __('Inactive: marked automatically after :period without logging in.', [
            'period' => $period,
        ]);

        if (timebank_config('profile_inactive.re-activate_at_login')) {
            $inactive .= ' ' . __('Logging in again removes the mark.');
        }

        // Closing a profile is removal, not inactivity: DeleteUser sets
        // deleted_at on the profile, never inactive_at. Those profiles are
        // absent from this chart, which is worth stating — the two states are
        // easy to conflate.
        $inactive .= ' ' . __('Removed profiles are not counted: removal is a separate state.');

        return [
            $inactive,
            __('Balance: what those profiles hold at each month\'s end, on the right-hand axis when there is any.'),
            __('Incomplete: too little filled in under description, tags, languages or location. Completing the profile removes it from the count.')
                // The state only started being recorded when incomplete_at was
                // added; anything earlier was estimated from the registration
                // date, so the chart should not present it as measured.
                . ' ' . __('Dates before the column was introduced are estimated from the registration date.'),
        ];
    }

    /**
     * How many profiles in the selection stand incomplete right now.
     *
     * Counts a set incomplete_at rather than re-deriving the rule, so the tile
     * and the chart can never disagree.
     */
    private function incompleteNowCount(): int
    {
        $count = 0;

        foreach ($this->profileKeys() as $type => $ids) {
            if (empty($ids) || $type === Bank::class) {
                continue;
            }

            $table = (new $type)->getTable();

            if (!\Schema::hasColumn($table, 'incomplete_at')) {
                continue;
            }

            $count += DB::table($table)
                ->whereIn('id', $ids)
                ->whereNotNull('incomplete_at')
                ->where('incomplete_at', '<=', now())
                ->count();
        }

        return $count;
    }

    /**
     * The month each profile was recorded incomplete, keyed 'Y-m'.
     *
     * Reads incomplete_at, which profiles:mark-incomplete maintains against the
     * live rule (ProfileTrait::hasIncompleteProfile()). Before that column
     * existed this had to be a present-day count with no history; now it is a
     * real series, accumulated by the caller exactly like the inactivity marks.
     *
     * Banks carry the column for consistency but are deliberately excluded:
     * a bank has no profile text to fill in, so counting it would only add a
     * constant.
     *
     * @return array<string, int>
     */
    private function incompleteByMonth(): array
    {
        $byMonth = [];

        foreach ($this->profileKeys() as $type => $ids) {
            if (empty($ids) || $type === Bank::class) {
                continue;
            }

            $table = (new $type)->getTable();

            if (!\Schema::hasColumn($table, 'incomplete_at')) {
                continue;
            }

            $rows = DB::table($table)
                ->whereIn('id', $ids)
                ->whereNotNull('incomplete_at')
                ->where('incomplete_at', '<=', now())
                ->selectRaw("DATE_FORMAT(incomplete_at, '%Y-%m') AS ym, COUNT(*) AS cnt")
                ->groupBy('ym')
                ->get();

            foreach ($rows as $row) {
                $byMonth[$row->ym] = ($byMonth[$row->ym] ?? 0) + (int) $row->cnt;
            }
        }

        return $byMonth;
    }

    /**
     * Net balance change per month across the accounts of profiles marked
     * inactive, in minutes. Keyed 'Y-m'.
     *
     * Same CASE expression as balance(): a transfer between two accounts in the
     * set nets to zero, so internal movement does not inflate the total. The
     * per-month deltas are summed into a running balance by the caller — a
     * balance is a position, not a period quantity.
     *
     * Only live accounts count. A removed profile has had its currency
     * withdrawn (DeleteUser creates a "Currency removal" transaction for the
     * full balance), so including one would chart hours that no longer exist.
     *
     * @param array<class-string, array<int>> $profileKeys
     * @return array<string, int>
     */
    private function inactiveBalanceByMonth(array $profileKeys): array
    {
        $profileKeys = array_filter($profileKeys, fn($ids) => !empty($ids));

        if (empty($profileKeys)) {
            return [];
        }

        $accountIds = Account::notRemoved()
            ->where(function ($q) use ($profileKeys) {
                foreach ($profileKeys as $type => $ids) {
                    $q->orWhere(function ($w) use ($type, $ids) {
                        $w->where('accountable_type', $type)->whereIn('accountable_id', $ids);
                    });
                }
            })
            ->pluck('id')
            ->map('intval')
            ->all();

        if (empty($accountIds)) {
            return [];
        }

        $in = $this->inList($accountIds);

        if ($in === null) {
            return [];
        }

        return DB::table('transactions')
            ->whereRaw("(from_account_id IN ($in) OR to_account_id IN ($in))")
            ->selectRaw("
                DATE_FORMAT(created_at, '%Y-%m') AS ym,
                SUM(CASE
                    WHEN to_account_id IN ($in) AND from_account_id IN ($in) THEN 0
                    WHEN to_account_id IN ($in) THEN amount
                    ELSE -amount
                END) AS delta
            ")
            ->groupBy('ym')
            ->get()
            ->pluck('delta', 'ym')
            ->map(fn($d) => (int) $d)
            ->all();
    }

    /**
     * How many profiles stood inactive at the end of each month.
     *
     * Cumulative, not per-month. Inactivity marks arrive in bursts — a single
     * migration marked 992 profiles inactive in one month — so a per-month
     * series is one spike beside years of zero, which says nothing about how
     * the inactive population actually developed. The running total answers
     * the question the section is asking; the month's own additions stay
     * available per row for the tooltip.
     *
     * Each row carries one key per profile type plus 'count' (the combined
     * running total), matching buildRegistrationsTimeline's shape, and
     * 'balance' — the hours those profiles held at that month's end.
     *
     * @param array<string, int> $balanceByMonth Net balance change per month,
     *                                           in minutes, keyed 'Y-m'.
     * @param array<string, int> $incompleteByMonth How many profiles were
     *                           recorded incomplete each month, keyed 'Y-m'.
     *                           Accumulated like the inactivity marks.
     */
    private function buildInactiveTimeline(
        array $byType,
        array $balanceByMonth = [],
        array $incompleteByMonth = []
    ): array {
        // The two series move on different months: a transaction can land in a
        // month where nobody was marked inactive, and vice versa. The axis has
        // to span both, or one line would be cut short.
        $months = collect($byType)->flatMap(fn($series) => array_keys($series))
            ->merge(array_keys($balanceByMonth))
            ->merge(array_keys($incompleteByMonth))
            ->unique();

        if ($months->isEmpty()) {
            return [];
        }

        $rows  = $months->mapWithKeys(fn($ym) => [$ym => (object) ['ym' => $ym]])->sortKeys();
        $types = array_keys($byType);

        // zeroFillMonths clamps the axis to the filter period, so marks
        // that happened before it opens are off-chart. Without seeding them the
        // line would restart at zero and understate every month shown.
        $start = $this->filter->periodFrom?->copy()->startOfMonth()?->format('Y-m');
        $running = [];
        foreach ($types as $type) {
            $running[$type] = $start === null ? 0 : collect($byType[$type])
                ->filter(fn($count, $ym) => $ym < $start)
                ->sum();
        }

        // Same seeding for the balance: it is a position carried forward from
        // before the window, not a quantity that starts at zero each period.
        $runningBalance = $start === null ? 0 : collect($balanceByMonth)
            ->filter(fn($delta, $ym) => $ym < $start)
            ->sum();

        // And for the incomplete count, which is cumulative for the same
        // reason: profiles recorded incomplete before the window still are.
        $runningIncomplete = $start === null ? 0 : collect($incompleteByMonth)
            ->filter(fn($count, $ym) => $ym < $start)
            ->sum();

        // A month with no new marks must still carry the running total
        // forward, so the empty row cannot be a constant — it is resolved per
        // month by the same closure.
        $carry = function () use (&$running, &$runningBalance, &$runningIncomplete, $types) {
            $values = [];
            foreach ($types as $type) {
                $values[$type] = $running[$type];
            }
            $values['count'] = array_sum($running);
            $values['added'] = 0;
            // Charted in hours, matching every other currency axis.
            $values['balance'] = round($runningBalance / 60, 2);
            $values['incomplete'] = $runningIncomplete;
            return $values;
        };

        return $this->zeroFillMonths(
            $rows,
            function ($row) use ($byType, $types, $balanceByMonth, $incompleteByMonth, &$running, &$runningBalance, &$runningIncomplete) {
                $values = [];
                $added  = 0;
                foreach ($types as $type) {
                    $running[$type] += $byType[$type][$row->ym] ?? 0;
                    $added          += $byType[$type][$row->ym] ?? 0;
                    $values[$type]   = $running[$type];
                }
                $values['count'] = array_sum($running);
                // The month's own additions, for the tooltip.
                $values['added'] = $added;

                $runningBalance += $balanceByMonth[$row->ym] ?? 0;
                // Charted in hours, matching every other currency axis.
                $values['balance'] = round($runningBalance / 60, 2);
                $runningIncomplete += $incompleteByMonth[$row->ym] ?? 0;
                $values['incomplete'] = $runningIncomplete;
                return $values;
            },
            $carry
        );
    }

    /**
     * Merge per-type registration months into continuous rows.
     *
     * Each row carries one key per profile type plus 'count' (the combined
     * total), so consumers can chart the split or use the aggregate.
     */
    private function buildRegistrationsTimeline(array $byType): array
    {
        $months = collect($byType)->flatMap(fn($series) => array_keys($series))->unique();

        if ($months->isEmpty()) {
            return [];
        }

        $rows = $months->mapWithKeys(fn($ym) => [$ym => (object) ['ym' => $ym]])->sortKeys();
        $types = array_keys($byType);

        $empty = array_fill_keys($types, 0) + ['count' => 0];

        return $this->zeroFillMonths(
            $rows,
            function ($row) use ($byType, $types) {
                $values = [];
                foreach ($types as $type) {
                    $values[$type] = $byType[$type][$row->ym] ?? 0;
                }
                $values['count'] = array_sum($values);
                return $values;
            },
            $empty
        );
    }

    /** Posts, calls, reactions and chat messages produced by the profile set. */
    public function engagement(): array
    {
        $keys = array_filter($this->profileKeys(), fn($ids) => !empty($ids));

        if (empty($keys)) {
            return [
                'posts' => 0, 'calls' => 0, 'reactions' => 0, 'messages' => 0,
                'timeline' => [],
            ];
        }

        $monthly = [
            'posts'     => $this->monthlyPolymorphicCounts('posts', 'postable_type', 'postable_id', $keys),
            'calls'     => $this->monthlyPolymorphicCounts('calls', 'callable_type', 'callable_id', $keys),
            'messages'  => $this->monthlyPolymorphicCounts('wirechat_messages', 'sendable_type', 'sendable_id', $keys),
            'reactions' => $this->monthlyReactionCounts($keys),
        ];

        return [
            // Period totals for the summary blocks.
            'posts'     => array_sum($monthly['posts']),
            'calls'     => array_sum($monthly['calls']),
            'messages'  => array_sum($monthly['messages']),
            'reactions' => array_sum($monthly['reactions']),
            // One row per month, carrying every series — shares the x-axis with
            // the other charts.
            'timeline'  => $this->buildEngagementTimeline($monthly),
        ];
    }

    /**
     * Counts per month for a polymorphic table, keyed by "Y-m". One query.
     *
     * @return array<string,int>
     */
    private function monthlyPolymorphicCounts(string $table, string $typeColumn, string $idColumn, array $keys): array
    {
        $query = $this->applyPeriod(DB::table($table))
            ->where(function ($q) use ($keys, $typeColumn, $idColumn) {
                foreach ($keys as $type => $ids) {
                    $q->orWhere(function ($w) use ($type, $ids, $typeColumn, $idColumn) {
                        $w->where($typeColumn, $type)->whereIn($idColumn, $ids);
                    });
                }
            });

        if (\Schema::hasColumn($table, 'deleted_at')) {
            $query->whereNull('deleted_at');
        }

        return $query->selectRaw("DATE_FORMAT(created_at, '%Y-%m') AS ym, COUNT(*) AS cnt")
            ->groupBy('ym')
            ->pluck('cnt', 'ym')
            ->map(fn($c) => (int) $c)
            ->all();
    }

    /** @return array<string,int> */
    private function monthlyReactionCounts(array $keys): array
    {
        $reacterIds = $this->reacterIds($keys);

        if ($reacterIds->isEmpty()) {
            return [];
        }

        return $this->applyPeriod(DB::table('love_reactions'))
            ->whereIn('reacter_id', $reacterIds)
            ->selectRaw("DATE_FORMAT(created_at, '%Y-%m') AS ym, COUNT(*) AS cnt")
            ->groupBy('ym')
            ->pluck('cnt', 'ym')
            ->map(fn($c) => (int) $c)
            ->all();
    }

    /**
     * Merge the per-series month maps into continuous rows for the chart.
     */
    private function buildEngagementTimeline(array $monthly): array
    {
        $months = collect($monthly)->flatMap(fn($series) => array_keys($series))->unique();

        if ($months->isEmpty()) {
            return [];
        }

        $rows = $months->mapWithKeys(fn($ym) => [$ym => (object) ['ym' => $ym]])->sortKeys();

        return $this->zeroFillMonths(
            $rows,
            fn($row) => [
                'posts'     => $monthly['posts'][$row->ym] ?? 0,
                'calls'     => $monthly['calls'][$row->ym] ?? 0,
                'messages'  => $monthly['messages'][$row->ym] ?? 0,
                'reactions' => $monthly['reactions'][$row->ym] ?? 0,
            ],
            ['posts' => 0, 'calls' => 0, 'messages' => 0, 'reactions' => 0]
        );
    }

    /**
     * Laravel Love reacter ids for the resolved profile set.
     */
    private function reacterIds(array $keys): \Illuminate\Support\Collection
    {
        $reacterIds = collect();

        foreach ($keys as $type => $ids) {
            $table = (new $type)->getTable();
            if (!\Schema::hasColumn($table, 'love_reacter_id')) {
                continue;
            }
            $reacterIds = $reacterIds->merge(
                DB::table($table)->whereIn('id', $ids)->whereNotNull('love_reacter_id')->pluck('love_reacter_id')
            );
        }

        return $reacterIds;
    }

    /** Most active profiles, and totals grouped by location. */
    public function distribution(): array
    {
        $in = $this->accountIn();

        if ($in === null) {
            return ['top_profiles' => collect(), 'by_location' => collect(), 'excluded_profiles' => []];
        }

        // Each transaction is attributed to the profiles on BOTH sides. Expressed
        // as a UNION ALL of two equality joins rather than one OR-join: an
        // `a.id = x.from OR a.id = x.to` join condition cannot use an index and
        // degrades into a nested-loop scan (measured at ~21s on 34k rows).
        $sides = $this->sideJoin('from_account_id', $in)
            ->unionAll($this->sideJoin('to_account_id', $in));

        // With a counterparty filter the selected profile is, by construction,
        // party to every transaction counted here — it would top both tables
        // and say nothing. Drop its own rows while keeping its transactions,
        // which are exactly what the partners' figures describe.
        $excluded = $this->excludedProfileKeys();
        $excludeRows = function ($query) use ($excluded) {
            foreach ($excluded as $type => $ids) {
                $query->whereNot(function ($w) use ($type, $ids) {
                    $w->where('s.type', $type)->whereIn('s.pid', $ids);
                });
            }

            return $query;
        };

        // Paid and received are reported separately: a profile that sends a lot
        // and one that receives a lot are both "active", but they are not the
        // same thing, and a single combined figure hides which is which.
        $top = $excludeRows(DB::query()->fromSub($sides, 's'))
            ->selectRaw('s.type, s.pid')
            ->selectRaw('COUNT(DISTINCT s.tx_id) AS cnt')
            ->selectRaw('COUNT(DISTINCT s.partner) AS partners')
            ->selectRaw('SUM(s.amount) AS volume')
            ->selectRaw("COUNT(DISTINCT CASE WHEN s.direction = 'paid' THEN s.tx_id END) AS paid_count")
            ->selectRaw("COUNT(DISTINCT CASE WHEN s.direction = 'received' THEN s.tx_id END) AS received_count")
            ->selectRaw("COALESCE(SUM(CASE WHEN s.direction = 'paid' THEN s.amount END), 0) AS paid_volume")
            ->selectRaw("COALESCE(SUM(CASE WHEN s.direction = 'received' THEN s.amount END), 0) AS received_volume")
            ->groupBy('s.type', 's.pid')
            // Ordering happens in SQL, before the limit: sorting the fetched
            // twenty would only reshuffle them, never surface the actual top
            // payer. Both values come off an allow-list in StatsFilter.
            ->orderBy($this->filter->topSortColumn(), $this->filter->topSortDirection())
            ->limit(20)
            ->get();

        $names = $this->profileNames($top);


        $topProfiles = $top->map(fn($r) => [
            'type'            => class_basename($r->type),
            'id'              => (int) $r->pid,
            'name'            => $names[$r->type . ':' . $r->pid] ?? __('Unknown'),
            // Identity-based, so the graph's labels match this table even
            // though the two are built over different transaction sets.
            'anonymous_name'  => self::pseudonym($r->type . ':' . $r->pid)
                ?? ($names[$r->type . ':' . $r->pid] ?? __('Unknown')),
            'count'           => (int) $r->cnt,
            // Distinct counterparties - the same measure the graph calls degree.
            'partners'        => (int) $r->partners,
            'volume'          => (int) $r->volume,
            'paid_count'      => (int) $r->paid_count,
            'received_count'  => (int) $r->received_count,
            'paid_volume'     => (int) $r->paid_volume,
            'received_volume' => (int) $r->received_volume,
        ]);

        // Both sides, split by direction — the same shape as the top-profiles
        // table above, so the two sections answer the question the same way.
        //
        // The paid and received columns each count a transaction once, so they
        // are the figures that reconcile with the section totals. The combined
        // columns count a transaction for both ends: that is what makes them
        // "activity touching this place", and it is also why they can exceed
        // the section total when both parties sit in the selection.
        //
        // LEFT join: a profile with no location row at all must still be
        // counted, under "not specified", or the rows silently stop adding up.
        // An inner join here quietly dropped 22 payments.
        $byLocation = $excludeRows(DB::query()->fromSub($sides, 's'))
            ->leftJoin('locations as l', function ($j) {
                $j->on('l.locatable_type', '=', 's.type')
                  ->on('l.locatable_id', '=', 's.pid');
            })
            ->selectRaw('l.country_id, l.division_id, l.city_id')
            ->selectRaw('COUNT(DISTINCT s.tx_id) AS cnt')
            ->selectRaw('SUM(s.amount) AS volume')
            ->selectRaw("COUNT(DISTINCT CASE WHEN s.direction = 'paid' THEN s.tx_id END) AS paid_count")
            ->selectRaw("COUNT(DISTINCT CASE WHEN s.direction = 'received' THEN s.tx_id END) AS received_count")
            ->selectRaw("COALESCE(SUM(CASE WHEN s.direction = 'paid' THEN s.amount END), 0) AS paid_volume")
            ->selectRaw("COALESCE(SUM(CASE WHEN s.direction = 'received' THEN s.amount END), 0) AS received_volume")
            ->groupBy('l.country_id', 'l.division_id', 'l.city_id')
            // Sorted in SQL before the limit, for the same reason as the
            // profiles table: ordering the fetched rows would only reshuffle
            // them. Both values come off the allow-list in StatsFilter.
            ->orderBy($this->filter->locationSortColumn(), $this->filter->locationSortDirection())
            ->limit(50)
            ->get();

        return [
            'top_profiles'      => $topProfiles,
            'by_location'       => $this->labelLocations($byLocation),
            'excluded_profiles' => $this->excludedProfileNames(),
        ];
    }

    /**
     * One side of the transaction/account relationship as an equality join, so
     * MySQL can use the index on that column. Both sides are UNIONed by callers.
     */
    private function sideJoin(string $column, string $in)
    {
        // The opposite side of the transaction, so a caller can count distinct
        // trading partners. Joined here rather than in a second pass: this
        // subquery already walks every matching row.
        $other = $column === 'from_account_id' ? 'to_account_id' : 'from_account_id';

        $query = $this->applyPeriod(DB::table('transactions as x'), 'x.created_at')
            ->join('accounts as a', 'a.id', '=', "x.$column")
            ->leftJoin('accounts as o', 'o.id', '=', "x.$other")
            ->whereRaw("a.id IN ($in)");

        // Same restriction as scopeToSet(): with a counterparty filter, only
        // transactions involving the selected profiles count.
        $counterpartyIds = $this->counterpartyAccountIds();
        if (!empty($counterpartyIds)) {
            $cin = $this->inList($counterpartyIds);
            $query->whereRaw("(x.from_account_id IN ($cin) OR x.to_account_id IN ($cin))");
        }

        // The direction tag lets callers split paid from received without a
        // second pass: the UNION halves are otherwise indistinguishable.
        $direction = $column === 'from_account_id' ? 'paid' : 'received';

        return $query->selectRaw(
            "a.accountable_type AS type, a.accountable_id AS pid, x.id AS tx_id, x.amount AS amount, '$direction' AS direction, "
            // NULL when the counterparty is the profile itself, so a transfer
            // between a profile's own accounts is not counted as a trading
            // partner. COUNT(DISTINCT ...) skips NULLs.
            . "CASE WHEN o.accountable_type = a.accountable_type AND o.accountable_id = a.accountable_id "
            . "THEN NULL ELSE CONCAT(o.accountable_type, ':', o.accountable_id) END AS partner"
        );
    }

    /**
     * Total balance held by the profile set.
     *
     * Cumulative by definition, so no period filter. Internal transfers net to
     * zero here — unlike volume, this is the case where `THEN 0` is correct.
     *
     * With a counterparty filter this reports the balance of the *selected*
     * profiles only. The wider set is "everyone they traded with", and summing
     * those balances would answer a question nobody asked — filtering on one
     * member would otherwise report the combined holdings of every organisation
     * and bank they ever transacted with.
     */
    /**
     * How current balances are spread across the profiles in the set.
     *
     * A histogram is the conventional way to show the distribution of a
     * quantity across a population, and it's the honest choice here: it answers
     * "are balances clustered near zero or spread into deep credit and debt?"
     * directly. A Lorenz curve / Gini coefficient — the other standard tool —
     * assumes non-negative values, and timebank balances are routinely negative,
     * so it would be misleading.
     *
     * Bands are fixed rather than evenly spaced because the real distribution is
     * heavily skewed: most profiles sit within a few hours of zero while a
     * currency-issuing bank can sit thousands of hours negative. Even-width
     * buckets would put almost every profile in one bar.
     *
     * Balances are cumulative by definition, so no period filter is applied —
     * the period still decides which profiles qualify.
     *
     * @return array{bands: array<int,array{label:string,count:int}>, profiles_counted:int}
     */
    public function balanceDistribution(): array
    {
        // With a counterparty filter, chart the trading partners — the selected
        // profile's own balance is already the headline figure.
        $accountIds = $this->distributionAccountIds();

        if (empty($accountIds)) {
            return ['bands' => [], 'profiles_counted' => 0];
        }

        $in = $this->inList($accountIds);

        // Banks are excluded: a bank issues and withdraws currency, so its
        // balance is the mirror image of everyone else's and lands in the
        // outermost band every time. Charting it would flatten the bands that
        // describe how members actually sit.
        $charted = array_values(array_diff(StatsFilter::PROFILE_TYPES, [Bank::class]));
        $typeIn = implode(',', array_map(fn($t) => "'" . addslashes($t) . "'", $charted));

        // One row per profile. The join matches each transaction to whichever of
        // the profile's accounts it touched, so a transfer between two accounts
        // of the *same* profile produces one +amount and one -amount row and
        // nets out naturally — no CASE needed, and transfers between different
        // profiles are left intact.
        $balances = DB::table('accounts as a')
            ->join('transactions as x', function ($j) {
                $j->on('x.from_account_id', '=', 'a.id')->orOn('x.to_account_id', '=', 'a.id');
            })
            ->whereRaw("a.id IN ($in)")
            ->whereRaw("a.accountable_type IN ($typeIn)")
            ->groupBy('a.accountable_type', 'a.accountable_id')
            ->selectRaw("a.accountable_type AS type, SUM(CASE WHEN x.to_account_id = a.id THEN x.amount ELSE -x.amount END) / 60 AS hours")
            ->get()
            ->map(fn($row) => ['type' => $row->type, 'hours' => (float) $row->hours]);

        if ($balances->isEmpty()) {
            return ['bands' => [], 'series' => [], 'profiles_counted' => 0];
        }

        // The balances are already in hand, so the observed extremes cost
        // nothing and describe the axis better than the limits alone.
        $observed = $balances->pluck('hours')->sort()->values();
        // A high percentile, not the maximum: the band width should describe
        // where people actually are, not where the single largest holder is.
        $typical = $observed->isEmpty()
            ? null
            : (float) $observed[(int) floor(0.95 * ($observed->count() - 1))];

        $bands = $this->balanceBands(
            $accountIds, $charted, $observed->min(), $observed->max(), $typical
        );

        // One row of counts per charted profile type, plus a combined total that
        // keeps the existing `count` key meaningful for the table beside it.
        $counts = [];
        foreach ($charted as $type) {
            $counts[$type] = array_fill(0, count($bands), 0);
        }

        foreach ($balances as $entry) {
            $hours = $entry['hours'];
            $type  = $entry['type'];

            if (!isset($counts[$type])) {
                continue;
            }

            foreach ($bands as $i => $band) {
                // The exact-zero band is its own bucket so "square" profiles are
                // visible rather than hidden inside a surrounding range.
                if ($band['min'] === 0 && $band['max'] === 0) {
                    if ($hours == 0) { $counts[$type][$i]++; break; }
                    continue;
                }
                $aboveMin = $band['min'] === null || $hours > $band['min'];
                $belowMax = $band['max'] === null || $hours <= $band['max'];
                if ($aboveMin && $belowMax && $hours != 0) { $counts[$type][$i]++; break; }
            }
        }

        $totals = array_map(
            fn($i) => collect($counts)->sum(fn($row) => $row[$i]),
            array_keys($bands)
        );

        // Rounding the axis up to a whole step can leave several empty bands
        // past the last occupied one. Keep exactly one — enough to show that the
        // range ends there — and drop the rest, which are only dead width.
        //
        // The low end is left alone: balanceBands() already starts it at zero or
        // at the lowest balance, so it has at most the one intended pad.
        $last = count($bands) - 1;
        while ($last > 0 && $totals[$last] === 0 && $totals[$last - 1] === 0) {
            $last--;
        }

        $keep = range(0, $last);

        return [
            // min/max travel with each band so consumers can reason about the
            // range numerically instead of parsing the formatted label.
            'bands' => collect($keep)
                ->map(fn($i) => [
                    'label' => $bands[$i]['label'],
                    'range' => $bands[$i]['range'] ?? $bands[$i]['label'],
                    'count' => $totals[$i],
                    'min'   => $bands[$i]['min'],
                    'max'   => $bands[$i]['max'],
                ])
                ->values()
                ->all(),
            'series' => collect($charted)
                ->mapWithKeys(fn($type) => [
                    class_basename($type) => array_values(array_intersect_key($counts[$type], array_flip($keep))),
                ])
                ->all(),
            'profiles_counted' => $balances->count(),
        ];
    }

    /**
     * Uniform bands spanning the balances actually present.
     *
     * Every band is the same width and states both of its bounds, so a reader
     * can see where the extremes really stop. An open-ended "> 50" cannot: it
     * reads the same whether the profiles inside it hold 51 hours or 5.100.
     *
     * The range comes from the observed balances, not the configured account
     * limits: a limit is what an account *may* reach, and charting to it leaves
     * empty bands that flatten the ones carrying people. Limits are still
     * consulted for the zero-floor case below.
     *
     * One empty band is added at each end, so it is always visible that the
     * occupied range stops where it does rather than running off the axis.
     *
     * @param  array<int>          $accountIds
     * @param  array<class-string> $charted
     * @return array<array{label: string, min: float|null, max: float|null}>
     */
    private function balanceBands(
        array $accountIds,
        array $charted,
        ?float $lowest = null,
        ?float $highest = null,
        ?float $typical = null
    ): array {
        if ($lowest === null || $highest === null) {
            return [$this->zeroBand()];
        }

        $in     = $this->inList($accountIds);
        $typeIn = implode(',', array_map(fn($t) => "'" . addslashes($t) . "'", $charted));

        $limits = DB::table('accounts')
            ->whereRaw("id IN ($in)")
            ->whereRaw("accountable_type IN ($typeIn)")
            ->selectRaw('MIN(limit_min) AS min_limit')
            ->first();

        // A platform whose accounts cannot go negative keeps the axis at zero
        // even if rounding would otherwise dip below it.
        $floorAtZero = $limits && ((int) $limits->min_limit) >= 0 && $lowest >= 0;

        $step = $this->bandStep($lowest, $highest, $typical);

        // Snap the ends outward to whole steps, then pad one empty band each
        // side so the edges of the occupied range are visible.
        $from = (floor($lowest / $step) - 1) * $step;
        $to   = (ceil($highest / $step) + 1) * $step;

        if ($floorAtZero) {
            $from = 0.0;
        }

        $bands = [];

        // Edges are computed as index * step rather than by accumulating
        // $edge += $step. Accumulation drifts: stepping up from -2.0 by 0.2
        // lands on -2.78e-16 instead of 0, which produced a stray zero-width
        // band and made the bands non-uniform.
        $edgeAt = fn(int $i, float $origin) => round($origin + ($i * $step), 9);

        // Zero is its own bucket: "square" profiles are a meaningful group and
        // would otherwise hide inside the first positive band.
        if ($from <= 0 && $to >= 0) {
            $negativeCount = (int) ceil(abs($from) / $step);
            for ($i = 0; $i < $negativeCount; $i++) {
                $lower = $edgeAt($i, $from);
                $upper = min($edgeAt($i + 1, $from), 0.0);

                // A band that has collapsed onto zero is the drift case above,
                // and would duplicate the exact-zero bucket.
                if ($upper <= $lower) {
                    continue;
                }

                $bands[] = $this->rangeBand($lower, $upper);
            }

            $bands[] = $this->zeroBand();

            $positiveCount = (int) ceil($to / $step);
            for ($i = 0; $i < $positiveCount; $i++) {
                $bands[] = $this->rangeBand($edgeAt($i, 0.0), $edgeAt($i + 1, 0.0));
            }
        } else {
            $count = (int) ceil(($to - $from) / $step);
            for ($i = 0; $i < $count; $i++) {
                $bands[] = $this->rangeBand($edgeAt($i, $from), $edgeAt($i + 1, $from));
            }
        }

        return $bands;
    }

    /**
     * A round band width that splits the range into a readable number of bands.
     *
     * Balances are long-tailed, which pulls the step in two directions: sizing
     * on the maximum crushes the bulk into one band, sizing on a percentile
     * gives the tail a long run of empty bands. Since every band is closed and
     * the tail must still be shown, the span comes from the maximum and the
     * band count is raised instead — that keeps the bulk spread over several
     * bands without stretching the axis into mostly-empty ones.
     *
     * $typical (a high percentile) only guards the reverse case: a single
     * extreme outlier far beyond everyone else, where the maximum alone would
     * make every real balance share one band.
     *
     * Steps stay on the 1/2/5 scale so the labels are numbers a person would
     * choose themselves (10, 25, 50) rather than artefacts of the data (13.7).
     */
    private function bandStep(float $lowest, float $highest, ?float $typical = null): float
    {
        // Only defer to the percentile when the tail is extreme enough that the
        // maximum would otherwise swallow the whole distribution.
        $upper = ($typical !== null && $typical > 0 && $highest > $typical * 10)
            ? $typical * 2
            : $highest;

        $span   = max($upper - min($lowest, 0.0), abs($upper), 1.0);
        // ~24 bands across the range. Doubled from 12 on request: at a step of
        // 10 hours the dense part of the distribution collapsed into two or
        // three bars, and halving the step resolves the bulk into roughly twice
        // as many occupied bands without changing what any band means.
        $target = $span / 24;

        $magnitude = 10 ** floor(log10(max($target, 0.01)));

        // 2.5 is on the ladder because without it a span just over 100 jumps
        // straight from a step of 10 to 20, halving the detail in the range
        // where most balances sit.
        //
        // The 5% tolerance matters more: a span of 121 h gives a target of
        // 10.1, which a strict `>=` rejects in favour of 20 — doubling every
        // band to accommodate a 1% overshoot. Accepting a candidate marginally
        // under the target costs one extra band and keeps the detail.
        foreach ([1, 2, 2.5, 5, 10] as $multiple) {
            $candidate = $multiple * $magnitude;
            if ($candidate >= $target * 0.95) {
                return (float) $candidate;
            }
        }

        return (float) (10 * $magnitude);
    }

    /**
     * A closed band, labelled with both of its bounds.
     *
     * The currency symbol appears once per label rather than on both bounds:
     * "H 100 - 110" carries the same meaning as "H 100 - H 110" in half the
     * width, which matters when fifteen of these share one axis.
     */
    private function rangeBand(float $min, float $max): array
    {
        // Both bounds. A lower bound alone ("H 0") reads as a point rather than
        // a span, which is precisely how the band starting at zero got mistaken
        // for the exact-zero bucket beside it. The currency symbol appears once,
        // so "H 0 - 5" stays no wider than it needs to be.
        $label = $this->bandBound($min) . ' - ' . $this->bandBound($max, false);

        return [
            'label' => $label,
            'range' => $label,
            'min'   => $min,
            'max'   => $max,
        ];
    }

    /**
     * The exact-zero bucket: profiles holding precisely nothing.
     *
     * Marked with "=" because it is a single value, not a span: every other
     * band is labelled with both bounds, so the equals sign is what separates
     * "exactly nothing" from "between nothing and five hours".
     *
     * @return array{label: string, range: string, min: int, max: int}
     */
    private function zeroBand(): array
    {
        // "= H 0" is unambiguous now that every other band shows both of its
        // bounds ("H 0 - 5"): an equals sign against a range reads clearly as a
        // single value. It was only confusing while the neighbouring band was
        // labelled "H 0" too.
        $label = '= ' . $this->bandBound(0);

        return ['label' => $label, 'range' => $label, 'min' => 0, 'max' => 0];
    }

    /**
     * A band bound in the platform's currency notation, e.g. "H 1.500".
     *
     * Follows the convention the other charts' axis ticks use rather than
     * tbFormat() directly: whole hours drop the ":00" (an axis reads better as
     * "H 100" than "H 100:00"), but a bound with real minutes still shows them,
     * so no value is ever displayed as something it isn't.
     */
    private function bandBound(float|int $value, bool $withSymbol = true): string
    {
        $value    = (float) $value;
        $negative = $value < 0;
        $absolute = abs($value);

        $wholeHours = (int) floor($absolute);
        $minutes    = (int) round(($absolute - $wholeHours) * 60);

        // Guard the 59.6 -> :60 rounding case.
        if ($minutes === 60) {
            $wholeHours++;
            $minutes = 0;
        }

        $separator = platform_trans('platform_currency_thousand_separator', null, '');
        $grouped = $separator !== ''
            ? number_format($wholeHours, 0, '', $separator)
            : (string) $wholeHours;

        $time = ($negative ? '-' : '') . $grouped
            . ($minutes === 0 ? '' : ':' . sprintf('%02d', $minutes));

        if (!$withSymbol) {
            return $time;
        }

        $symbol = platform_currency_symbol();

        return platform_trans('platform_currency_position_end', null, false)
            ? $time . ' ' . $symbol
            : $symbol . ' ' . $time;
    }

    /**
     * Profiles the distribution tables leave out, keyed by model class.
     *
     * Only the explicitly selected counterparties: they appear in every
     * transaction the filter counts, so listing them alongside their partners
     * compares a profile against its own dealings.
     *
     * @return array<class-string, array<int>>
     */
    private function excludedProfileKeys(): array
    {
        $keys = [];

        foreach ($this->filter->counterparties as $party) {
            $keys[$party['type']][] = (int) $party['id'];
        }

        return $keys;
    }

    /** Names of the excluded profiles, for the note shown above the tables. */
    public function excludedProfileNames(): array
    {
        $names = [];

        foreach ($this->excludedProfileKeys() as $type => $ids) {
            foreach ($type::whereIn('id', $ids)->pluck('name') as $name) {
                $names[] = $name;
            }
        }

        return $names;
    }

    /**
     * Accounts to chart in the balance distribution.
     *
     * Without a counterparty filter that's the whole resolved set. With one, it
     * is the trading partners — the selected profile is reported separately as
     * the headline balance, so including it here would just repeat that.
     *
     * @return array<int>
     */
    private function distributionAccountIds(): array
    {
        $all = $this->accountIds();
        $selected = $this->counterpartyAccountIds();

        if (empty($selected)) {
            return $all;
        }

        return array_values(array_diff($all, $selected));
    }

    public function balance(): int
    {
        $counterpartyIds = $this->counterpartyAccountIds();

        $in = !empty($counterpartyIds)
            ? $this->inList($counterpartyIds)
            : $this->accountIn();

        if ($in === null) {
            return 0;
        }

        $row = DB::table('transactions')
            ->whereRaw("(from_account_id IN ($in) OR to_account_id IN ($in))")
            ->selectRaw("
                SUM(CASE
                    WHEN to_account_id IN ($in) AND from_account_id IN ($in) THEN 0
                    WHEN to_account_id IN ($in) THEN amount
                    ELSE -amount
                END) AS balance
            ")
            ->first();

        return (int) ($row->balance ?? 0);
    }

    /**
     * The whole time-lapse: one graph, plus per step what changed in each of
     * its three layers.
     *
     * STEPS. 'month': month ends, the last one being $till itself. 'week':
     * every 7 days back from $till, so the last frame is exactly the end date.
     *
     * LAYERS, all measured in MONTHS whatever the step, so a weekly time-lapse
     * looks like the monthly one, only moving in smaller steps:
     *
     *   current  per month: that calendar month, up to the step's date
     *            per week:  the month ending on the step's date (rolling)
     *            accumulated: everything up to the step's date
     *   ghost    the month before `current` (not when accumulating)
     *   history  everything before that (not when accumulating)
     *
     * Each layer is a window sliding forward over the trade, grouped by day, so
     * one query and one pass serve every step. Only what enters or leaves a
     * layer at each step is returned - `frames`, `ghosts` and `history` are
     * lists of {nodes_added, nodes_removed, edges_added, edges_removed}, the
     * first entry of each being the full layer. Full layers per step would be
     * ~1.8 million ids for ten years of weeks; changes stay small.
     *
     * WHY ONE PASS. One network() per step runs an all-time query each time
     * (~1.1s a frame, 42.5s for 36 months before anything was drawn). Here the
     * graph is built once, for the layout, and the layers come from a single
     * grouped query.
     *
     * $till is an EXACT, inclusive moment - not rounded up to its month end.
     *
     * @return array{graph: array, steps: array, frames: array, ghosts: ?array, history: ?array, timeline: array}
     */
    public function networkTimelapse(
        Carbon $from,
        Carbon $till,
        string $step = 'month',
        bool $cumulative = true,
        int $minDegree = 0,
        ?array $scope = null,
        bool $edge = false,
        ?array $types = null,
        ?int $maxDegree = null,
        bool $anonymise = false
    ): array {
        $step = $step === 'week' ? 'week' : 'month';
        $last = $till->copy();
        $empty = ['graph' => ['nodes' => [], 'edges' => [], 'stats' => []], 'steps' => [],
                  'frames' => [], 'ghosts' => null, 'history' => null, 'volumes' => [],
                  'timeline' => ['paid' => [], 'removed' => [], 'inactive' => []]];

        // The union: everything traded up to the end. Every layer of every
        // step is a subset of this, so it is laid out once.
        $graph = (new self($this->filter->withPeriod(null, $last)))
            ->network($minDegree, $scope, $edge, $types, $maxDegree, $anonymise);

        $in = $this->accountIn();

        if (empty($graph['nodes']) || $in === null) {
            return ['graph' => $graph] + $empty;
        }

        // Only elements that survived the graph's own filtering may appear: a
        // node dropped for degree must not come back through a raw pair row.
        $keepNodes = array_flip(array_column($graph['nodes'], 'id'));
        $keepEdges = [];

        foreach ($graph['edges'] as $e) {
            $keepEdges[$e['source'] . '|' . $e['target'] . '|' . $e['type_id']] = true;
        }

        // Trade by DAY: fine enough for weekly steps, coarse enough to stay a
        // few tens of thousands of rows. Per side, whether it paid that day -
        // the same definition of "paid" computeNetwork() uses.
        $rows = DB::table('transactions as x')
            ->join('accounts as af', 'af.id', '=', 'x.from_account_id')
            ->join('accounts as at', 'at.id', '=', 'x.to_account_id')
            ->whereRaw("x.from_account_id IN ($in)")
            ->whereRaw("x.to_account_id IN ($in)")
            ->whereRaw("CONCAT(af.accountable_type, af.accountable_id) <> CONCAT(at.accountable_type, at.accountable_id)")
            ->whereRaw($this->graphTypesSql('x'))
            ->where('x.created_at', '<=', $last)
            ->selectRaw("
                DATE(x.created_at) AS d,
                LEAST(CONCAT(af.accountable_type, ':', af.accountable_id),
                      CONCAT(at.accountable_type, ':', at.accountable_id)) AS a,
                GREATEST(CONCAT(af.accountable_type, ':', af.accountable_id),
                         CONCAT(at.accountable_type, ':', at.accountable_id)) AS b,
                " . $this->canonicalTypeSql('x') . " AS type_id,
                SUM(x.amount) AS volume,
                MAX(CASE WHEN CONCAT(af.accountable_type, ':', af.accountable_id)
                    = LEAST(CONCAT(af.accountable_type, ':', af.accountable_id),
                            CONCAT(at.accountable_type, ':', at.accountable_id))
                    THEN 1 ELSE 0 END) AS a_paid,
                MAX(CASE WHEN CONCAT(af.accountable_type, ':', af.accountable_id)
                    <> LEAST(CONCAT(af.accountable_type, ':', af.accountable_id),
                             CONCAT(at.accountable_type, ':', at.accountable_id))
                    THEN 1 ELSE 0 END) AS b_paid
            ")
            ->groupBy('d', 'a', 'b', 'type_id')
            ->orderBy('d')
            ->get();

        // The steps, and each step's layer boundaries as 'Y-m-d' (inclusive).
        $steps = $this->timelapseSteps($from, $last, $step);

        if (empty($steps)) {
            return ['graph' => $graph] + $empty;
        }

        // Each profile's total traded time AS OF each step, sent only for the
        // steps where it changes - the volume view sizes nodes by it, so a
        // node grows as the time-lapse moves on instead of showing its final
        // all-time size from the first frame. (The partner count other views
        // size by is derived in the browser from the layers it already has.)
        //
        // Totalled straight into the steps while the day-ordered rows stream
        // past: a per-day list of [day, profile, minutes] put ~14 MB on the
        // peak for all of history in weeks.
        $volumes = array_fill(0, count($steps), []);
        $volumeTotal = [];
        $volumeStep = 0;
        $stepCount = count($steps);

        // Day-ordered events per kind: [day, key], once per key per day.
        $nodeEvents = [];
        $edgeEvents = [];
        $paidEvents = [];
        $seen = [];

        foreach ($rows as $r) {
            $day = (string) $r->d;

            // Trade before the first step counts towards the first step.
            while ($volumeStep < $stepCount - 1 && $day > $steps[$volumeStep]['till']) {
                $volumeStep++;
            }

            foreach ([$r->a, $r->b] as $id) {
                if (isset($keepNodes[$id])) {
                    $volumeTotal[$id] = ($volumeTotal[$id] ?? 0) + (int) $r->volume;
                    $volumes[$volumeStep][$id] = $volumeTotal[$id];
                }
            }

            foreach ([[$r->a, $r->a_paid], [$r->b, $r->b_paid]] as [$id, $paid]) {
                if (!isset($keepNodes[$id])) {
                    continue;
                }

                if (!isset($seen["n|$day|$id"])) {
                    $seen["n|$day|$id"] = true;
                    $nodeEvents[] = [$day, $id];
                }

                if ($paid && !isset($seen["p|$day|$id"])) {
                    $seen["p|$day|$id"] = true;
                    $paidEvents[] = [$day, $id];
                }
            }

            $key = $r->a . '|' . $r->b . '|' . $r->type_id;

            if (isset($keepEdges[$key]) && !isset($seen["e|$day|$key"])) {
                $seen["e|$day|$key"] = true;
                $edgeEvents[] = [$day, $key];
            }
        }

        unset($seen, $rows);

        $current = [new SlidingWindow($nodeEvents), new SlidingWindow($edgeEvents)];
        $ghost   = $cumulative ? null : [new SlidingWindow($nodeEvents), new SlidingWindow($edgeEvents)];
        $history = $cumulative ? null : [new SlidingWindow($nodeEvents), new SlidingWindow($edgeEvents)];

        $frames = [];
        $ghosts = $cumulative ? null : [];
        $back   = $cumulative ? null : [];

        $delta = fn (array $pair, ?string $start, string $end) => [
            ...$this->renameDelta($pair[0]->slideTo($start, $end), 'nodes'),
            ...$this->renameDelta($pair[1]->slideTo($start, $end), 'edges'),
        ];

        // Payment dates, for "last paid as of this step": the latest payment up
        // to the first step, then each step's latest - so weekly steps get
        // weekly precision, not a month-end date that lies in their future.
        $paidDates = [];
        $paidPointer = 0;
        $paidCount = count($paidEvents);

        foreach ($steps as $k => $s) {
            $frames[] = ['label' => $s['label'], 'till' => $s['till']]
                + $delta($current, $cumulative ? null : $s['current_from'], $s['till']);

            if (!$cumulative) {
                $ghosts[] = $delta($ghost, $s['ghost_from'], $s['ghost_till']);
                $back[]   = $delta($history, null, $s['history_till']);
            }

            $latest = [];

            while ($paidPointer < $paidCount && $paidEvents[$paidPointer][0] <= $s['till']) {
                [$day, $payer] = $paidEvents[$paidPointer++];
                $latest[$payer] = $day;
            }

            foreach ($latest as $payer => $day) {
                $paidDates[$payer][] = $day;
            }
        }

        // Objects even when empty, so the browser reads every step alike.
        $volumes = array_map(fn ($changed) => (object) $changed, $volumes);

        $dates = $this->profileDates(array_keys($keepNodes));

        return [
            'graph'    => $graph,
            'steps'    => array_map(fn ($s) => ['label' => $s['label'], 'till' => $s['till']], $steps),
            'frames'   => $frames,
            'ghosts'   => $ghosts,
            'history'  => $back,
            'volumes'  => $volumes,
            // Dated facts, so every step is coloured as of its own date rather
            // than today.
            'timeline' => [
                'paid'     => $paidDates,
                'removed'  => $dates['removed'],
                'inactive' => $dates['inactive'],
            ],
        ];
    }

    /**
     * The steps of a time-lapse and each one's layer boundaries, as 'Y-m-d'.
     *
     * Month: the calendar month is `current`, the month before is the ghost.
     * Week: `current` is the month ENDING on the step's date, so it slides by a
     * week each step; the ghost is the month before that. Either way history
     * is everything older than the ghost.
     *
     * @return array<array{label:string, till:string, current_from:string, ghost_from:string, ghost_till:string, history_till:string}>
     */
    private function timelapseSteps(Carbon $from, Carbon $last, string $step): array
    {
        $tills = [];

        if ($step === 'week') {
            $start = $from->copy()->startOfMonth()->startOfDay();

            for ($t = $last->copy()->startOfDay(); $t->gte($start); $t->subDays(7)) {
                $tills[] = $t->copy();
            }

            $tills = array_reverse($tills);
        } else {
            for ($t = $from->copy()->startOfMonth(); $t->lte($last); $t->addMonth()) {
                $end = $t->copy()->endOfMonth();
                $tills[] = ($end->gt($last) ? $last->copy() : $end)->startOfDay();
            }
        }

        return array_map(function (Carbon $t) use ($step) {
            $currentFrom = $step === 'week'
                ? $t->copy()->subMonthNoOverflow()->addDay()
                : $t->copy()->startOfMonth();
            $ghostFrom = $currentFrom->copy()->subMonthNoOverflow();

            return [
                'label'        => $t->format($step === 'week' ? 'j M Y' : 'M Y'),
                'till'         => $t->toDateString(),
                'current_from' => $currentFrom->toDateString(),
                'ghost_from'   => $ghostFrom->toDateString(),
                'ghost_till'   => $currentFrom->copy()->subDay()->toDateString(),
                'history_till' => $ghostFrom->copy()->subDay()->toDateString(),
            ];
        }, $tills);
    }

    /** {added, removed} from a SlidingWindow as {<kind>_added, <kind>_removed}. */
    private function renameDelta(array $delta, string $kind): array
    {
        return [
            $kind . '_added'   => $delta['added'],
            $kind . '_removed' => $delta['removed'],
        ];
    }

    /**
     * When each profile was removed and when it was marked inactive, as
     * 'Y-m-d', for the profiles that have either.
     *
     * The dated counterpart of profileStates(), which answers only "as of
     * now". Both states can change over the span of a time-lapse, so a frame
     * needs the date to decide for itself.
     *
     * @param array<string> $keys 'Type:id' profile keys.
     * @return array{removed: array<string,string>, inactive: array<string,string>}
     */
    private function profileDates(array $keys): array
    {
        $byType = [];

        foreach ($keys as $key) {
            [$type, $id] = explode(':', $key, 2);
            $byType[$type][] = (int) $id;
        }

        $removed = [];
        $inactive = [];

        foreach ($byType as $type => $ids) {
            if (!class_exists($type)) {
                continue;
            }

            foreach (array_chunk($ids, 1000) as $chunk) {
                $rows = $type::withoutGlobalScopes()
                    ->whereIn('id', $chunk)
                    ->where(fn ($q) => $q->whereNotNull('deleted_at')->orWhereNotNull('inactive_at'))
                    ->get(['id', 'deleted_at', 'inactive_at']);

                foreach ($rows as $row) {
                    $key = $type . ':' . $row->id;

                    if ($row->deleted_at !== null) {
                        $removed[$key] = Carbon::parse($row->deleted_at)->toDateString();
                    }

                    if ($row->inactive_at !== null) {
                        $inactive[$key] = Carbon::parse($row->inactive_at)->toDateString();
                    }
                }
            }
        }

        return ['removed' => $removed, 'inactive' => $inactive];
    }

    /**
     * Trading network: profiles as nodes, trading relationships as edges.
     *
     * This is *volume*, not balance: every transaction counts once and internal
     * transfers are NOT zeroed. See StatsAggregationTest — copying balance()'s
     * `THEN 0` rule here produces plausible but wrong numbers.
     *
     * @param int $minDegree Nodes with fewer trading partners drop out. On dev
     *                       the median degree is 2 and 46% of edges are
     *                       one-offs, so without a threshold the graph is
     *                       unreadable. Note that degree 0 cannot occur — a
     *                       node exists because of an edge — so 0 and 1 are
     *                       equivalent.
     * @param int|null $maxDegree Upper bound, to leave the hubs out. null (or
     *                       >= self::DEGREE_CEILING) means no bound.
     * @param array<string>|null $scope Keys ('App\Models\User:161') the graph is
     *                       limited to. null = no limit (Admin). For a bank:
     *                       its clients.
     * @param bool $edge Also show edges with one end outside the scope; the
     *                   counterparty then appears without a name.
     * @param array<string>|null $types Profile types to include ('User',
     *                   'Organization', 'Bank'). null = all.
     */
    public function network(
        int $minDegree = 3,
        ?array $scope = null,
        bool $edge = false,
        ?array $types = null,
        ?int $maxDegree = null,
        bool $anonymise = false
    ): array {
        // Cached on the full argument set, not just the filter: the same
        // filter with a different degree range or profile-type selection is a
        // different graph. Measured at ~900ms uncached on 1173 nodes / 2403
        // edges, and every server-side control rebuilds it.
        //
        // Locale is in the key because the payload carries translated
        // transaction-type names on every edge.
        $key = 'stats:net:' . self::CACHE_VERSION . ':' . app()->getLocale() . ':'
             . md5(serialize([
                 $this->filter->hash(),
                 $minDegree,
                 $scope,
                 $edge,
                 $types,
                 $maxDegree,
                 // Part of the key: an anonymised payload must never be served
                 // from a cache entry built with real names, or the other way
                 // round.
                 $anonymise,
             ]));

        return Cache::remember(
            $key,
            self::CACHE_TTL,
            fn() => $this->computeNetwork($minDegree, $scope, $edge, $types, $maxDegree, $anonymise)
        );
    }

    /**
     * The uncached aggregation behind network(). Split out so the cache wrapper
     * stays readable; see network() for what the arguments mean.
     */
    private function computeNetwork(
        int $minDegree = 3,
        ?array $scope = null,
        bool $edge = false,
        ?array $types = null,
        ?int $maxDegree = null,
        bool $anonymise = false
    ): array {
        $in = $this->accountIn();

        if ($in === null) {
            return ['nodes' => [], 'edges' => [], 'stats' => $this->networkStats(0, 0, 0, $minDegree)];
        }

        // One GROUP BY over two equality joins on accounts. Deliberately not a
        // self-join on transactions: when measured that had not finished after
        // 180s, and per sideJoin() an OR-join on transactions degrades into a
        // nested-loop scan of ~21s.
        $pairs = $this->applyPeriod(DB::table('transactions as x'), 'x.created_at')
            ->join('accounts as af', 'af.id', '=', 'x.from_account_id')
            ->join('accounts as at', 'at.id', '=', 'x.to_account_id')
            ->whereRaw("x.from_account_id IN ($in)")
            ->whereRaw("x.to_account_id IN ($in)")
            ->whereRaw("CONCAT(af.accountable_type, af.accountable_id) <> CONCAT(at.accountable_type, at.accountable_id)")
            // Rounding corrections and reversals only; currency creation,
            // removal and migration stay in so the graph shows where hours
            // enter and leave the network.
            ->whereRaw($this->graphTypesSql('x'))
            ->selectRaw("
                LEAST(CONCAT(af.accountable_type, ':', af.accountable_id),
                      CONCAT(at.accountable_type, ':', at.accountable_id)) AS a,
                GREATEST(CONCAT(af.accountable_type, ':', af.accountable_id),
                         CONCAT(at.accountable_type, ':', at.accountable_id)) AS b,
                COUNT(*) AS tx,
                SUM(x.amount) AS volume,
                -- Grouped on below: a pair that traded Work AND Donation
                -- becomes two edges, so each arrow carries one type and that
                -- type's own volume. 27% of pairs use more than one type.
                " . $this->canonicalTypeSql('x') . " AS ttype,
                SUM(CASE WHEN CONCAT(af.accountable_type, ':', af.accountable_id)
                    = LEAST(CONCAT(af.accountable_type, ':', af.accountable_id),
                            CONCAT(at.accountable_type, ':', at.accountable_id))
                    THEN x.amount ELSE 0 END) AS a_to_b,
                -- Last time each side PAID. The circulation view uses this to
                -- decide whether a profile has stopped passing hours on: a
                -- node that once circulated but has not paid anyone for months
                -- is stagnating, which pure topology cannot see.
                MAX(CASE WHEN CONCAT(af.accountable_type, ':', af.accountable_id)
                    = LEAST(CONCAT(af.accountable_type, ':', af.accountable_id),
                            CONCAT(at.accountable_type, ':', at.accountable_id))
                    THEN x.created_at END) AS a_last_paid,
                MAX(CASE WHEN CONCAT(af.accountable_type, ':', af.accountable_id)
                    <> LEAST(CONCAT(af.accountable_type, ':', af.accountable_id),
                             CONCAT(at.accountable_type, ':', at.accountable_id))
                    THEN x.created_at END) AS b_last_paid
            ")
            ->groupBy('a', 'b', 'ttype')
            ->get();

        if ($pairs->isEmpty()) {
            return ['nodes' => [], 'edges' => [], 'stats' => $this->networkStats(0, 0, 0, $minDegree)];
        }

        // Type filter before counting degrees: a filtered-out neighbour must
        // not count towards the threshold of a node that stays.
        if ($types !== null && $types !== []) {
            $allowed = array_flip($types);
            $ofType = fn(string $key) => isset($allowed[class_basename(explode(':', $key, 2)[0])]);

            $pairs = $pairs->filter(fn($pair) => $ofType($pair->a) && $ofType($pair->b))->values();

            if ($pairs->isEmpty()) {
                return ['nodes' => [], 'edges' => [], 'stats' => $this->networkStats(0, 0, 0, $minDegree)];
            }
        }

        // Scoping: for a bank only edges with at least one client end remain.
        // Without $edge BOTH ends must be clients.
        if ($scope !== null) {
            $inScope = array_flip($scope);

            $pairs = $pairs->filter(function ($pair) use ($inScope, $edge) {
                $a = isset($inScope[$pair->a]);
                $b = isset($inScope[$pair->b]);

                return $edge ? ($a || $b) : ($a && $b);
            })->values();

            if ($pairs->isEmpty()) {
                return ['nodes' => [], 'edges' => [], 'stats' => $this->networkStats(0, 0, 0, $minDegree)];
            }
        }

        // Count degrees in PHP: cheaper than a second SQL aggregation. Paid and
        // received come from the directions the edge already carries, so no
        // extra query is needed.
        $degree = [];
        $volume = [];
        $paid = [];
        $received = [];

        // Degree counts distinct PARTNERS, not rows. The pair query now
        // groups by transaction type too, so a pair that traded Work and
        // Donation yields two rows - counting rows would report two trading
        // partners where there is one.
        $seenPair = [];
        $lastPaid = [];

        $notePaid = function ($key, $when) use (&$lastPaid) {
            if ($when === null) {
                return;
            }

            if (!isset($lastPaid[$key]) || $when > $lastPaid[$key]) {
                $lastPaid[$key] = $when;
            }
        };

        foreach ($pairs as $pair) {
            $aToB = (int) $pair->a_to_b;
            $bToA = (int) $pair->volume - $aToB;

            $notePaid($pair->a, $pair->a_last_paid ?? null);
            $notePaid($pair->b, $pair->b_last_paid ?? null);

            $pairKey = $pair->a . '|' . $pair->b;
            $firstTimeForPair = !isset($seenPair[$pairKey]);
            $seenPair[$pairKey] = true;

            foreach ([$pair->a, $pair->b] as $key) {
                if ($firstTimeForPair) {
                    $degree[$key] = ($degree[$key] ?? 0) + 1;
                }
                $volume[$key] = ($volume[$key] ?? 0) + (int) $pair->volume;
            }

            $paid[$pair->a]     = ($paid[$pair->a] ?? 0) + $aToB;
            $received[$pair->b] = ($received[$pair->b] ?? 0) + $aToB;

            $paid[$pair->b]     = ($paid[$pair->b] ?? 0) + $bToA;
            $received[$pair->a] = ($received[$pair->a] ?? 0) + $bToA;
        }

        $totalNodes = count($degree);
        $totalEdges = $pairs->count();

        // Nodes outside the scope escape the threshold: it measures how
        // connected someone is within the circle, and an outsider should not
        // count towards that. They only appear if attached to a visible edge.
        $inScope = $scope === null ? null : array_flip($scope);

        $ceiling = ($maxDegree === null || $maxDegree >= self::DEGREE_CEILING) ? PHP_INT_MAX : $maxDegree;

        $keep = array_filter(
            $degree,
            fn($d, $key) => ($inScope !== null && !isset($inScope[$key]))
                || ($d >= $minDegree && $d <= $ceiling),
            ARRAY_FILTER_USE_BOTH
        );

        // An outsider without a visible neighbour is useless: drop it when its
        // only edges ran to filtered-out nodes.
        if ($inScope !== null) {
            $linked = [];
            foreach ($pairs as $pair) {
                if (isset($keep[$pair->a]) && isset($keep[$pair->b])) {
                    $linked[$pair->a] = true;
                    $linked[$pair->b] = true;
                }
            }
            $keep = array_intersect_key($keep, $linked);
        }

        // Which of the kept profiles are closed or marked inactive.
        //
        // Always resolved, not gated on the display toggles: the circulation
        // view treats a flagged-inactive profile as having stopped, and that
        // must not depend on whether someone happened to switch on the
        // highlight. One query over the kept keys.
        $states = $this->profileStates(array_keys($keep));

        $typeNames = $this->transactionTypeNames();

        $edges = $pairs
            ->filter(fn($p) => isset($keep[$p->a]) && isset($keep[$p->b]))
            ->map(fn($p) => [
                'source'  => $p->a,
                'target'  => $p->b,
                'tx'      => (int) $p->tx,
                'volume'  => (int) $p->volume,
                // One edge per transaction type, so an arrow says what kind
                // of movement it is and carries that type's own volume.
                'type_id'   => (int) $p->ttype,
                'type_name' => $typeNames[$p->ttype] ?? __('Other'),
                // Both directions separately, so the flow view can draw
                // outbound and inbound as two lines: a single net arrow hides
                // whether a pair exchanged 100 hours or barely traded.
                'a_to_b'  => (int) $p->a_to_b,
                'b_to_a'  => (int) $p->volume - (int) $p->a_to_b,
                // Net direction, so the arrow follows the actual flow.
                'net'     => (int) $p->a_to_b - ((int) $p->volume - (int) $p->a_to_b),
            ])
            ->values()
            ->all();

        // Every node is named, including profiles outside the scope.
        //
        // NOTE: this is a deliberate loosening of the earlier rule, on the
        // platform operator's instruction. A bank now sees the names of members
        // who are not its clients; previously those were withheld. The visual
        // distinction (grey, dashed) and the bank label in the tooltip remain,
        // so it stays clear which network a node belongs to — but the identity
        // is no longer hidden. The anonymised variant
        // (networkAnonymised()) is unaffected and still carries no names.
        $named = array_keys($keep);

        $names   = $this->networkNames($named);
        $regions = $this->networkRegions(array_keys($keep));

        // The bank network each profile belongs to. Looked up for every node,
        // not just outsiders: the admin graph shows it in the tooltip too.
        $bankLabels = $this->networkBankNames(array_keys($keep));

        $nodes = [];
        foreach ($keep as $key => $deg) {
            [$type, $id] = explode(':', $key, 2);
            $outside = $inScope !== null && !isset($inScope[$key]);

            $nodes[] = [
                'id'      => $key,
                // Scoping only ever applies to a bank (an Admin gets null), so
                // naming the bank here is accurate wherever this appears.
                // Identity is resolved HERE, not in the browser. Sending the
                // real name with an "anonymise" flag and letting JS choose put
                // every member's name in the page source, readable by anyone
                // who opened DevTools - the toggle looked like privacy but was
                // only presentation.
                'name'      => $anonymise
                    ? (self::pseudonym($key) ?? $names[$key]['name'] ?? class_basename($type) . ' ' . $id)
                    : ($names[$key]['name'] ?? class_basename($type) . ' ' . $id),
                // Never sent while anonymising: it is the one field that
                // defeats a pseudonym outright.
                'full_name' => $anonymise ? null : ($names[$key]['full_name'] ?? null),
                // Which bank network this profile belongs to, for the tooltip.
                // Filled for every node, so the admin graph can show it too.
                'bank'    => $bankLabels[$key] ?? null,
                // Closed by its owner: drawn as a dashed outline with no fill.
                'removed'  => (bool) ($states[$key]['removed'] ?? false),
                // Marked inactive: normal fill, blue outline.
                'inactive' => (bool) ($states[$key]['inactive'] ?? false),
                'type'    => class_basename($type),
                'degree'  => $deg,
                'volume'   => $volume[$key] ?? 0,
                // Days since this profile last paid anyone, or null if it
                // never has. The circulation view compares this against its
                // "flow stopped after" slider.
                'days_since_paid' => isset($lastPaid[$key])
                    ? (int) (\Carbon\Carbon::parse($lastPaid[$key])->diffInDays(now(), true))
                    : null,
                'paid'     => $paid[$key] ?? 0,
                'received' => $received[$key] ?? 0,
                'outside' => $outside,
                'region'  => $regions[$key] ?? __('Location unknown'),
            ];
        }

        usort($nodes, fn($x, $y) => $y['volume'] <=> $x['volume']);

        // The most recent trade that is actually DRAWN: the latest payment on
        // any edge that survived the filters. Exports state it instead of
        // "All time", which says nothing about when the figure was made. Every
        // transaction is one side's payment, so the pairs' last-paid dates
        // cover them all.
        $lastDate = $pairs
            ->filter(fn($p) => isset($keep[$p->a]) && isset($keep[$p->b]))
            ->flatMap(fn($p) => [$p->a_last_paid ?? null, $p->b_last_paid ?? null])
            ->filter()
            ->max();

        return [
            'nodes' => $nodes,
            'edges' => $edges,
            'stats' => $this->networkStats($totalNodes, $totalEdges, count($nodes), $minDegree)
                + [
                    'max_degree'     => $maxDegree,
                    'highest_degree' => max($degree ?: [0]),
                    'last_date'      => $lastDate ? substr((string) $lastDate, 0, 10) : null,
                ],
        ];
    }

    /**
     * The bank each profile belongs to, by profile key.
     *
     * Shown in the node tooltip, so a reader can see which network a trading
     * partner belongs to. No profile belongs to two banks (verified on the
     * data), so one name per key is unambiguous; profiles with no bank at all
     * are simply absent from the map.
     *
     * @param array<string> $keys
     * @return array<string, string>
     */
    private function networkBankNames(array $keys): array
    {
        if (empty($keys)) {
            return [];
        }

        $byType = [];
        foreach ($keys as $key) {
            [$type, $id] = explode(':', $key, 2);
            $byType[$type][] = (int) $id;
        }

        $rows = collect();
        foreach ($byType as $type => $ids) {
            $rows = $rows->merge(
                DB::table('bank_clients')
                    ->where('client_type', $type)
                    ->whereIn('client_id', $ids)
                    ->get(['bank_id', 'client_type', 'client_id'])
            );
        }

        if ($rows->isEmpty()) {
            return [];
        }

        $bankNames = DB::table('banks')
            ->whereIn('id', $rows->pluck('bank_id')->unique())
            ->pluck('name', 'id');

        $labels = [];
        foreach ($rows as $row) {
            $name = $bankNames[$row->bank_id] ?? null;

            if ($name === null) {
                continue;
            }

            // The plain bank name: it is shown as a labelled tooltip row now,
            // not as a node label, so ":bank network" would read as a tautology.
            $labels[$row->client_type . ':' . $row->client_id] ??= $name;
        }

        return $labels;
    }

    /**
     * Region per node, for the region view.
     *
     * Two things keep this honest:
     *
     * - Profiles without a location go into a visible "unknown" group rather
     *   than being dropped. On dev 32% of users and 75% of organisations have
     *   no location; hiding those silently makes the map look more complete
     *   than it is.
     * - Regions with fewer than MIN_REGION_SIZE profiles are merged into
     *   "other". A city with two members is not a region but a hint
     *   (k-anonymity).
     *
     * @param array<string> $keys
     * @return array<string, string>
     */
    public const MIN_REGION_SIZE = 5;

    private function networkRegions(array $keys): array
    {
        $byType = [];
        foreach ($keys as $key) {
            [$type, $id] = explode(':', $key, 2);
            $byType[$type][] = (int) $id;
        }

        $rows = collect();
        foreach ($byType as $type => $ids) {
            $located = DB::table('locations')
                ->where('locatable_type', $type)
                ->whereIn('locatable_id', $ids)
                ->select('locatable_id', 'city_id', 'division_id', 'country_id')
                ->get()
                ->map(fn($r) => (object) [
                    'key'         => $type . ':' . $r->locatable_id,
                    'city_id'     => $r->city_id,
                    'division_id' => $r->division_id,
                    'country_id'  => $r->country_id,
                ]);

            $rows = $rows->merge($located);
        }

        if ($rows->isEmpty()) {
            return array_fill_keys($keys, __('Location unknown'));
        }

        // Deliberately not labelLocations(): that is built for the location
        // aggregation and expects count fields (cnt, paid_volume, ...) a node
        // row does not have. Only the short label is needed here.
        $locale    = app()->getLocale();
        $cities    = \App\Models\Locations\City::whereIn('id', $rows->pluck('city_id')->filter())->with('translations')->get()->keyBy('id');
        $divisions = \App\Models\Locations\Division::whereIn('id', $rows->pluck('division_id')->filter())->with('translations')->get()->keyBy('id');
        $countries = \App\Models\Locations\Country::whereIn('id', $rows->pluck('country_id')->filter())->with('translations')->get()->keyBy('id');

        $name = fn($model) => $model?->translations?->firstWhere('locale', $locale)?->name
            ?? $model?->translations?->first()?->name;

        $region = [];
        foreach ($rows as $row) {
            $city     = $name($cities[$row->city_id] ?? null);
            $division = $name($divisions[$row->division_id] ?? null);
            $country  = $name($countries[$row->country_id] ?? null);

            // The same short form the profile cards use: "City, Division".
            $label = $city
                ? ($division ? $city . ', ' . $division : $city)
                : ($division ?: ($country ?: __('Location unknown')));

            // A profile can have several locations; the first one wins.
            $region[$row->key] ??= $label;
        }

        // k-anonymity: merge regions that are too small.
        $counts = array_count_values($region);
        foreach ($region as $key => $label) {
            if (($counts[$label] ?? 0) < self::MIN_REGION_SIZE) {
                $region[$key] = __('Other regions');
            }
        }

        foreach ($keys as $key) {
            $region[$key] ??= __('Location unknown');
        }

        return $region;
    }

    /**
     * Anonymised graph, for a view shared with members.
     *
     * Deliberately a separate method rather than a flag on network(): a flag
     * can be set wrongly, two methods cannot. What comes out contains no names,
     * no profile ids and no exact amounts.
     *
     * Three measures, because anonymising alone is not enough for a graph:
     *
     * 1. Keys become a salted hash. The salt lives in APP_KEY and cannot be
     *    derived from the payload; without a salt a list of profile ids would
     *    be enough to reverse the hashes.
     * 2. Volume is banded rather than exact: an exact amount is a fingerprint
     *    that ties a node to a transaction.
     * 3. The threshold is a privacy requirement here, not a readability
     *    choice — a node with a single edge is identifiable in practice. Hence
     *    the floor of MIN_ANONYMOUS_DEGREE, even when the caller asks for less.
     */
    public const MIN_ANONYMOUS_DEGREE = 3;

    public function networkAnonymised(int $minDegree = self::MIN_ANONYMOUS_DEGREE): array
    {
        $graph = $this->network(max($minDegree, self::MIN_ANONYMOUS_DEGREE));

        $salt = config('app.key');
        $hash = fn(string $key) => 'n' . substr(hash_hmac('sha256', $key, $salt), 0, 10);

        // Volume bands from this graph's own quantiles: absolute thresholds
        // would land arbitrarily on a different platform.
        $volumes = array_values(array_filter(array_column($graph['nodes'], 'volume')));
        sort($volumes);
        $q = function (float $p) use ($volumes) {
            if (empty($volumes)) {
                return 0;
            }
            return $volumes[(int) floor(($volumes ? count($volumes) - 1 : 0) * $p)];
        };
        $q1 = $q(0.33);
        $q2 = $q(0.66);

        $band = function ($volume) use ($q1, $q2) {
            if ($volume <= $q1) {
                return 'low';
            }
            return $volume <= $q2 ? 'mid' : 'high';
        };

        $nodes = array_map(fn($node) => [
            'id'     => $hash($node['id']),
            // No name. The type stays: needed for colour, barely revealing.
            'type'   => $node['type'],
            'degree' => $node['degree'],
            'band'   => $band($node['volume']),
            // Region may travel: networkRegions() has already merged groups
            // that are too small, so this points at no individual.
            'region' => $node['region'],
        ], $graph['nodes']);

        $edges = array_map(fn($edge) => [
            'source' => $hash($edge['source']),
            'target' => $hash($edge['target']),
            // Transaction count stays (it drives line width), exact volume does not.
            'tx'     => $edge['tx'],
            // Direction only, not the amount: enough to draw the arrow without
            // disclosing a figure.
            'dir'    => $edge['net'] <=> 0,
        ], $graph['edges']);

        return [
            'nodes' => $nodes,
            'edges' => $edges,
            'stats' => $graph['stats'],
        ];
    }

    /** What the threshold left out, so the UI can report it honestly. */
    private function networkStats(int $nodes, int $edges, int $shown, int $minDegree): array
    {
        return [
            'total_nodes' => $nodes,
            'total_edges' => $edges,
            'shown_nodes' => $shown,
            'min_degree'  => $minDegree,
        ];
    }

    /**
     * Names for the nodes. Only called for in-scope keys; the anonymised
     * variant never reaches this method.
     *
     * @param array<string> $keys 'App\Models\User:161'
     * @return array<string, array{name:string, full_name:?string}>
     */
    private function networkNames(array $keys): array
    {
        $byType = [];
        foreach ($keys as $key) {
            [$type, $id] = explode(':', $key, 2);
            $byType[$type][] = (int) $id;
        }

        $names = [];
        foreach ($byType as $type => $ids) {
            if (!class_exists($type)) {
                continue;
            }
            // Both columns in the one query that was already being made: the
            // tooltip shows the full name under the username.
            $rows = $type::withoutGlobalScopes()
                ->whereIn('id', $ids)
                ->get(['id', 'name', 'full_name']);

            foreach ($rows as $row) {
                $names[$type . ':' . $row->id] = [
                    'name'      => $row->name,
                    // Suppressed when it merely repeats the username, so the
                    // tooltip does not show the same string twice.
                    'full_name' => $row->full_name !== $row->name ? $row->full_name : null,
                ];
            }
        }

        return $names;
    }

    /**
     * Stable pseudonym for a profile, e.g. "User 12".
     *
     * Derived from the profile's own identity, NOT from its rank in a result
     * set. That is the whole point: the most-active table and the network graph
     * are built over different transaction sets — the graph has its own period
     * override, degree thresholds and type filters — so a rank-based number
     * would mean "12th busiest in THIS list" and the two sides would disagree
     * about who "User 12" is. Keyed on identity, the label is the same
     * everywhere, so searching the graph for the name shown in the table finds
     * the matching profile.
     *
     * The number is the profile id, which is already visible to any admin who
     * can reach this page, and is per-type so the ids stay short and readable.
     *
     * This is display-level pseudonymisation for the operator's own screen, and
     * is deliberately NOT a privacy boundary: it is reversible by anyone with
     * database access. The shareable export is networkAnonymised(), which hashes
     * ids, bands volumes and enforces a degree floor. Do not use this in its
     * place.
     *
     * Banks are returned as null: a platform has a handful of them, so a number
     * conceals nothing while making the row harder to read. The caller falls
     * back to the real name.
     *
     * @param string $key 'App\Models\User:161'
     */
    public static function pseudonym(string $key): ?string
    {
        [$type, $id] = array_pad(explode(':', $key, 2), 2, '');
        $base = class_basename($type);

        if ($base === 'Bank' || $base === '' || $id === '') {
            return null;
        }

        return __($base) . ' ' . $id;
    }

    // ------------------------------------------------------------------ helpers

    /** Comma-separated account id list, or null when the set is empty. */
    private function accountIn(): ?string
    {
        $ids = $this->accountIds();

        return empty($ids) ? null : $this->inList($ids);
    }

    /** Ints only — this cast is what keeps the raw fragments injection-safe. */
    private function inList(array $ids): string
    {
        return implode(',', array_map('intval', $ids));
    }

    private function applyPeriod($query, string $column = 'created_at')
    {
        if ($this->filter->periodFrom) {
            $query->where($column, '>=', $this->filter->periodFrom->copy()->startOfDay());
        }
        if ($this->filter->periodTill) {
            $query->where($column, '<=', $this->filter->periodTill->copy()->endOfDay());
        }

        return $query;
    }

    /**
     * Transaction-type ids in the order they should stack, bottom first.
     *
     * Ids rather than names because the names are translated — ordering on the
     * display string would give a different stack per locale. Any type not
     * listed (a platform may add its own) keeps its id order after these, and
     * the "Unknown" bucket sits last since it is a fallback, not a type.
     */
    private const TYPE_STACK_ORDER = [
        1, // Work
        2, // Gift
        3, // Donation
        6, // Migration
        7, // Rounding correction
        8, // Reversal
        4, // Currency creation
        5, // Currency removal
    ];

    /**
     * Sort the type names present into TYPE_STACK_ORDER.
     *
     * @param  array<string> $names names actually present in the data
     * @return array<string>
     */
    private function orderedTypeNames(array $names): array
    {
        $typeNames = $this->transactionTypeNames();
        $rank = [];

        foreach (self::TYPE_STACK_ORDER as $position => $id) {
            if (isset($typeNames[$id])) {
                $rank[$typeNames[$id]] = $position;
            }
        }

        // Types the platform added beyond the known set follow, in id order.
        $next = count(self::TYPE_STACK_ORDER);
        foreach ($typeNames as $name) {
            if (!isset($rank[$name])) {
                $rank[$name] = $next++;
            }
        }

        $unknownRank = $next + 1; // "Unknown" is a fallback bucket, so it sits last.

        usort($names, fn($a, $b) => ($rank[$a] ?? $unknownRank) <=> ($rank[$b] ?? $unknownRank));

        return $names;
    }

    private function transactionTypeNames(): array
    {
        // Cache the RAW names and translate per call. rememberForever() used
        // to store the translated strings, so whichever locale warmed the
        // cache first was frozen for every other one - an English session saw
        // Dutch type names, permanently, because the cache never expires.
        $raw = Cache::rememberForever(
            'stats:transaction_type_names_raw',
            fn() => TransactionType::pluck('name', 'id')->all()
        );

        return collect($raw)
            ->map(fn($name) => __(ucfirst(strtolower($name))))
            ->all();
    }

    /**
     * Fill gaps between the first and last month so charts have a continuous axis.
     */
    /**
     * @param array|callable $empty Values for a month with no data. A callable
     *                              is resolved per month, which a cumulative
     *                              series needs so gaps carry the running total
     *                              forward instead of dropping to zero.
     */
    private function zeroFillMonths($rows, callable $map, array|callable $empty): array
    {
        if ($rows->isEmpty()) {
            return [];
        }

        $keys  = $rows->keys();
        $start = $this->filter->periodFrom?->copy()->startOfMonth()
            ?? \Carbon\Carbon::createFromFormat('Y-m', $keys->first())->startOfMonth();
        $end   = $this->filter->periodTill?->copy()->startOfMonth()
            ?? \Carbon\Carbon::createFromFormat('Y-m', $keys->last())->startOfMonth();

        $out = [];
        foreach (CarbonPeriod::create($start, '1 month', $end) as $month) {
            $ym = $month->format('Y-m');
            $out[] = array_merge(
                ['label' => $month->translatedFormat('M Y'), 'ym' => $ym],
                isset($rows[$ym]) ? $map($rows[$ym]) : (is_callable($empty) ? $empty() : $empty)
            );
        }

        return $out;
    }

    /** Resolve names for the handful of profiles in a top-N result. */
    private function profileNames($rows): array
    {
        $names = [];

        foreach ($rows->groupBy('type') as $type => $group) {
            if (!class_exists($type)) {
                continue;
            }
            $ids = $group->pluck('pid');
            foreach ($type::whereIn('id', $ids)->pluck('name', 'id') as $id => $name) {
                $names[$type . ':' . $id] = $name;
            }
        }

        return $names;
    }

    /** Attach translated city/division/country names to grouped location rows. */
    private function labelLocations($rows)
    {
        $cities    = \App\Models\Locations\City::whereIn('id', $rows->pluck('city_id')->filter())->with('translations')->get()->keyBy('id');
        $divisions = \App\Models\Locations\Division::whereIn('id', $rows->pluck('division_id')->filter())->with('translations')->get()->keyBy('id');
        $countries = \App\Models\Locations\Country::whereIn('id', $rows->pluck('country_id')->filter())->with('translations')->get()->keyBy('id');

        $name = fn($model) => $model?->translations?->firstWhere('locale', app()->getLocale())?->name
            ?? $model?->translations?->first()?->name;

        /**
         * The same short form the profile cards use (MainSearchBar::…
         * 'location_short'): "City, Division", falling back to
         * "Division, CC" and then the country code alone. The country is its
         * code rather than its name, which is what keeps the label short.
         */
        $short = function ($city, $division, $countryCode, $countryName) {
            if ($city) {
                return $division ? $city . ', ' . $division : $city;
            }

            if ($division) {
                return $countryCode ? $division . ', ' . $countryCode : $division;
            }

            // A country on its own reads better by name than by code, and one
            // of them is the "XX" placeholder whose name ("~ Location not
            // specified") is the whole point of the row.
            return $countryName ?: $countryCode ?: null;
        };

        return $rows->map(fn($r) => [
            'city'            => $name($cities[$r->city_id] ?? null),
            'division'        => $name($divisions[$r->division_id] ?? null),
            'country'         => $name($countries[$r->country_id] ?? null),
            'label'           => $short(
                $name($cities[$r->city_id] ?? null),
                $name($divisions[$r->division_id] ?? null),
                ($countries[$r->country_id] ?? null)?->code,
                $name($countries[$r->country_id] ?? null),
            ),
            'count'           => (int) $r->cnt,
            'volume'          => (int) $r->volume,
            'paid_count'      => (int) $r->paid_count,
            'received_count'  => (int) $r->received_count,
            'paid_volume'     => (int) $r->paid_volume,
            'received_volume' => (int) $r->received_volume,
        ]);
    }
}