<?php

namespace App\Helpers;

use App\Jobs\AutoDecisionJob;
use App\Jobs\DatapointNotificationJob;
use App\Jobs\RunRejectionJob;
use App\Services\RawRedisConnector;
use Illuminate\Support\Facades\DB;
use App\Logging\CloudWatchLoggerFactory;

class DatapointHelper
{
    /** One Redis hash per application: loan:{id}:datapoints */
    const KEY_PREFIX = 'loan:';
    const KEY_SUFFIX = ':datapoints';

    /** Hash TTL — refreshed on every write. */
    const TTL_SECONDS = 1800; // 30 minutes

    /**
     * Why a readiness notification went out.
     *
     * COLD_START — the hash did not exist when this write happened, so nothing
     * downstream has been told anything about this application's readiness yet.
     * Delivered by a direct in-process call, not through SQS.
     *
     * CHANGED — the hash existed and at least one tracked flag flipped.
     * Delivered through SQS, because this is the hot save path.
     */
    const NOTIFY_COLD_START = 'cold_start';
    const NOTIFY_CHANGED = 'changed';

    /**
     * AUTO_BRE flow switch. 'database' keeps the old RabbitMQ event flow;
     * anything else (including unset) runs this Redis/SQS flow. See
     * isRedisFlow().
     */
    const FLOW_DATABASE = 'mysql';
    const FLOW_REDIS = 'redis';

    /**
     * Feature go-live. Applications created BEFORE this are out of scope for the
     * whole tracker — see isEligible().
     *
     * Override with DATAPOINT_RELEASE_DATE (plain env(), not _env(): a date is
     * not a secret). Anything strtotime() can parse.
     *
     * Set this to the actual go-live timestamp before deploying.
     */
    const RELEASE_DATE = '2026-08-19 00:00:00';

    /**
     * The dispatch guards live in their OWN hash — loan:{id}:dispatch_guard —
     * not in the data point hash.
     *
     * They have to outlive the data points. The data point hash is a readiness
     * index with a short, rolling TTL; losing it costs nothing because it is
     * rebuilt from the database, and the documented backfill procedure deletes
     * it outright. A guard is the record that a job already went to SQS —
     * losing that means sending the job twice. Hence a separate key with a
     * long TTL, set once at claim time and never rolled forward.
     */
    const JOB_KEY_SUFFIX = ':dispatch_guard';

    /** Guard lifetime from the moment of dispatch. Long enough that no live
     *  application outlives it, bounded so the keys are not immortal. */
    const JOB_TTL_SECONDS = 7776000; // 90 days

    /**
     * Idempotency guard for the approval tier. The applicant type is appended,
     * so the applicant and the co-applicant can each dispatch exactly once.
     */
    const AUTODECISION_JOB_PREFIX = 'autodecision_dispatched_';

    /** Idempotency guard for the rejection tier. */
    const REJECTION_JOB_PREFIX = 'rejection_dispatched_';

    /**
     * The tiers, each one (a field list => a job) with its own guard.
     *
     * Array order is dispatch priority and it is deliberate: approval is
     * evaluated first because it subsumes rejection. The rejection field list
     * is a strict subset of the approval one, so an application that completes
     * approval has also completed rejection — and AutoDecisionJob runs the
     * rejection policy itself (generateSoftBureauReport -> runAutoDecision ->
     * autoApproveInit -> autoReject). Dispatching both would run that policy
     * twice, so an approval dispatch claims the rejection guard as well.
     *
     * The reverse never happens: a rejection dispatch leaves the approval guard
     * alone, so approval still fires later when its remaining 19 fields land.
     */
    public static $tiers = [
        'approval' => [
            'job' => AutoDecisionJob::class,
            'guard' => self::AUTODECISION_JOB_PREFIX,
            'subjob' => ['rejection'],
        ],
        'rejection' => [
            'job' => RunRejectionJob::class,
            'guard' => self::REJECTION_JOB_PREFIX,
            'subjob' => [],
        ],
    ];

    /**
     * All fields required for the approval rule to run.
     * This is the single source of truth for the approval field list.
     *
     * Except for consent_log (which means "an agreed consent row exists"),
     * every name here is a literal column name on consumer_application or
     * consumer_application_element. Verify against the schema before adding
     * one — a name that is not a real column makes !empty() permanently false,
     * pinning its flag to 0 so the tier can never complete.
     *
     * The credit report is deliberately absent: the bureau pull happens inside
     * AutoDecisionJob, so it is an output of decisioning, not a precondition.
     */
    public static $approvalRequired = [
        // consumer_application (rejection subset)
        'name', 'dob', 'pan_card', 'monthly_income', 'consent_log',
        // consumer_application
        'loan_amount', 'offer_id', 'company_name',
        'current_address_line_1', 'current_address_pincode',
        'current_address_state', 'current_address_city', 'employee_status',
        // consumer_application_element
        'product_name',
        'installation_address_line_1', 'installation_address_pincode',
        'installation_address_city', 'installation_address_state',
        'solar_power_type', 'type_of_customer',
        'inverter_monthly_power_bill', 'inverter_KW_installation',
        'inverter_gross_project_amount', 'inverter_power_savings',
    ];

    /**
     * All fields required for the rejection rule to run (subset of approval).
     */
    public static $rejectionRequired = [
        'name', 'dob', 'pan_card', 'monthly_income', 'consent_log',
    ];

    /**
     * Human-friendly labels for every tracked data point.
     *
     * Keys are the raw Redis / column names used throughout the helper.
     * Values are what the merchant or admin sees in notifications and the UI.
     * Any field not listed here falls back to ucwords(str_replace('_', ' ', $field)).
     */
    public static $labels = [
        // --- applicant core (rejection tier) ---
        'name'                          => 'Full Name',
        'dob'                           => 'Date of Birth',
        'pan_card'                      => 'PAN Card',
        'monthly_income'                => 'Monthly Income',
        'consent_log'                   => 'Applicant Consent',
        // --- applicant address & loan (approval tier) ---
        'loan_amount'                   => 'Loan Amount',
        'offer_id'                      => 'Offer',
        'company_name'                  => 'Company / Employer Name',
        'current_address_line_1'        => 'Current Address Line 1',
        'current_address_pincode'       => 'Current Address Pincode',
        'current_address_state'         => 'Current Address State',
        'current_address_city'          => 'Current Address City',
        'employee_status'               => 'Employment Status',
        // --- element / installation (approval tier) ---
        'product_name'                  => 'Product',
        'installation_address_line_1'   => 'Installation Address Line 1',
        'installation_address_pincode'  => 'Installation Address Pincode',
        'installation_address_city'     => 'Installation Address City',
        'installation_address_state'    => 'Installation Address State',
        'solar_power_type'              => 'Solar Power Type',
        'type_of_customer'              => 'Customer Type',
        'inverter_monthly_power_bill'   => 'Monthly Power Bill',
        'inverter_KW_installation'      => 'System Capacity (kW)',
        'inverter_gross_project_amount' => 'Gross Project Amount',
        'inverter_power_savings'        => 'Power Savings',
        // --- co-applicant (rejection tier) ---
        'co_applicant_name'             => 'Co-Applicant Full Name',
        'co_applicant_dob'              => 'Co-Applicant Date of Birth',
        'co_applicant_pan_card'         => 'Co-Applicant PAN Card',
        'co_applicant_mobile_number'    => 'Co-Applicant Mobile Number',
        'co_applicant_monthly_income'   => 'Co-Applicant Monthly Income',
        'co_applicant_address_pincode'  => 'Co-Applicant Address Pincode',
        'co_applicant_rejection_executed' => 'Co-Applicant Rejection Check Pending',
    ];

    /**
     * Return the human-friendly label for a data-point key.
     *
     * Falls back to title-casing the raw key so an unlabelled field is still
     * readable rather than silently missing.
     */
    public static function label(string $field): string
    {
        return self::$labels[$field] ?? ucwords(str_replace('_', ' ', $field));
    }

    /**
     * Translate a list of raw field names to their human-friendly labels.
     *
     * @param  string[] $fields
     * @return string[]
     */
    public static function labelled(array $fields): array
    {
        return array_map([self::class, 'label'], $fields);
    }

    /**
     * All fields required for the CO-APPLICANT rejection rule to run.
     *
     * Every name is a column on consumer_application. Two of them —
     * co_applicant_pan_card and co_applicant_mobile_number — are in
     * ConsumerApplication::$encryptable, so buildFromDatabase() (raw query
     * builder) reads ciphertext where the observer reads plaintext. That is
     * fine for a presence flag: Encryptable::setAttribute() only encrypts a
     * non-empty value, so a blank column stays NULL/'' on both paths and the
     * two agree.
     *
     * co_applicant_name is load-bearing beyond being required: it is also how
     * the whole feature decides whether a co-applicant exists at all. See
     * hasCoApplicant().
     *
     * NOTE ON THE PINCODE FIELD. The spec for this tier said
     * `co_applicant_pincode`. There is no such column — the schema has
     * `co_applicant_address_pincode` (2018_08_29_100238), matching the
     * co_applicant_address_* block in ConsumerApplication::$fillable. The real
     * name is used here; the spelled name would have made !empty() permanently
     * false and pinned the flag to 0, so the tier could never complete. This is
     * the same trap the first cut of the applicant list fell into.
     *
     * Consent is deliberately NOT in this list. Instead, co-applicant consent
     * gates *who may dispatch* the tier: only the consent document choke point
     * does (ConsentController@addConsentDocument), never the observer hot path.
     *
     * That is what keeps the tier from completing on the six identity fields at
     * form-completion time — long before consent — dispatching, and then dying:
     * generateSoftBureauReport() refuses to pull for applicant type 2 without a
     * co_app_consent_logs file or is_co_app_consent = 1 ('Please get
     * Co-Applicant consent.'), the job would return without deciding, and the
     * guard would already be claimed for 90 days with nothing to clear it.
     *
     * The ordering that makes the consent paths safe: saveConsent() sets
     * is_co_app_consent BEFORE saving the consent_log row, and it is that save
     * which triggers dispatch — so the flag the pull checks is already set.
     *
     * The cost of this choice: six fields completing AFTER consent do not
     * re-trigger on their own, since no other caller dispatches type 2.
     */
    public static $coApplicantRejectionRequired = [
        'co_applicant_name', 'co_applicant_dob', 'co_applicant_pan_card',
        'co_applicant_mobile_number', 'co_applicant_monthly_income',
        'co_applicant_address_pincode',
    ];

    public static function log(array $context)
    {
        (new CloudWatchLoggerFactory())->logger(json_encode($context));
    }

    /**
     * The field list for a tier and applicant type.
     *
     * An unsupported (tier, applicant type) combination returns an EMPTY list —
     * co-applicant approval, for instance, which is not wired up. Callers must
     * read empty as "this tier does not apply to this applicant", never as
     * "no fields are missing": the field loop in attemptDispatch() would
     * otherwise find nothing missing and dispatch immediately. That check lives
     * in exactly one place (attemptDispatch), so this stays the single
     * declaration of which combinations exist.
     */
    public static function required(string $tier = 'approval', int $applicantType = 1, array $datapoints = null): array
    {
        if (self::normaliseApplicantType($applicantType) === 2) {
            return $tier === 'rejection' ? self::$coApplicantRejectionRequired : [];
        }

        $fields = $tier === 'rejection' ? self::$rejectionRequired : self::$approvalRequired;

        if ($tier === 'approval' && ($datapoints === null || self::hasCoApplicant($datapoints))) {
            $fields[] = 'co_applicant_rejection_executed';
        }

        return $fields;
    }

    /**
     * Every data point the tracker follows — the union across all tiers and
     * both applicant types, deduped, in declaration order.
     *
     * buildFromDatabase() and flagsFrom() both derive from this rather than from
     * $approvalRequired, which is what lets a new tier add fields (as the
     * co-applicant tier does) without either of them being touched.
     */
    public static function tracked(): array
    {
        $fields = [];
        foreach (array_keys(self::$tiers) as $tier) {
            foreach ([1, 2] as $applicantType) {
                foreach (self::required($tier, $applicantType) as $field) {
                    $fields[$field] = true;
                }
            }
        }

        return array_keys($fields);
    }

    /**
     * Does this application have a co-applicant?
     *
     * The co-applicant is optional, so this is the gate that keeps the whole
     * co-applicant half of the feature dormant on a single-applicant
     * application: no type-2 tier is evaluated, no type-2 guard is claimed, and
     * the notification reports no co-applicant pending fields rather than
     * reporting all six as missing.
     *
     * Presence is defined as co_applicant_name being populated. That is the
     * same test the rest of the codebase uses (generateSoftBureauReport,
     * ConsumerApplication::saveCustomer, AutoDecisionJob), so the tracker and
     * the jobs it dispatches cannot disagree about whether a co-applicant
     * exists.
     */
    public static function hasCoApplicant(array $datapoints): bool
    {
        return !empty($datapoints['co_applicant_name']);
    }

    /**
     * Is the Redis/SQS flow active, or has AUTO_BRE been switched back to the
     * old database (RabbitMQ event) flow?
     *
     * Read with _env(), which DECRYPTS via wall_decrypt() — so the value in
     * .env must be the wall_encrypt() ciphertext of 'database', not the literal
     * word. A plaintext AUTO_BRE=database silently fails this comparison and
     * leaves the Redis flow on. See the convention note in the documentation.
     *
     * Anything other than 'database' means the Redis flow, matching the
     * comparison already in ConsentController — an unset AUTO_BRE is therefore
     * the new flow, not the old one.
     */
    public static function isRedisFlow(): bool
    {
        return _env('AUTO_BRE') !== self::FLOW_DATABASE;
    }

    /** Per-request memo, keyed by application and current state. */
    protected static $eligibility = [];

    /**
     * Is this application in scope for the feature at all?
     *
     * Only applications created on or after RELEASE_DATE are. Without this gate,
     * the first post-release save of ANY historical application would find no
     * Redis key, cold-start the tracker, and — for the very many old
     * applications whose 24 approval fields are long since complete — claim the
     * guard and dispatch AutoDecisionJob. That means a fresh bureau pull and a
     * BRE run on a loan that was decided months ago, plus a WhatsApp message to
     * the merchant group about an application nobody is working on. RunRejectionJob
     * is worse: it only returns early when the state is already 'rejected', so a
     * disbursed loan could be moved to 'rejected'.
     *
     * Gated here rather than per-caller so every entry point inherits it: write(),
     * refresh(), attemptDispatch(), pending() and pendingByApplicant(). sync() and
     * dispatchIfReady() are covered transitively.
     *
     * @param  int        $applicationId
     * @param  mixed      $createdAt   Pass the value if you already have it (the
     *                                application observer does) to skip the lookup.
     * @param  string|null $state      Pass the current state when already loaded.
     * @param  mixed      $merchantId  Pass the value if you already have it to skip
     *                                the lookup. Used to fetch the merchant's
     *                                category_id from the merchants table.
     * @return bool
     */
    public static function isEligible(int $applicationId, $createdAt = null, $state = null, $merchantId = null): bool
    {
        if ($createdAt === null || $state === null || $merchantId === null) {
            try {
                $consumerApplication = DB::table('consumer_application')->where('id', $applicationId)->first(['created_at', 'state', 'merchant_id']);
                $createdAt = $consumerApplication->created_at;
                $state = $consumerApplication->state;
                $merchantId = $consumerApplication->merchant_id;
            } catch (\Exception $e) {
                return false;
            }
        }

        try {
            $merchant = DB::table('merchant')->where('id', $merchantId)->first(['category_id']);
            if (!$merchant || (string) $merchant->category_id !== (string) _env('SOLAR_CATEGORY_ID')) {
                return false;
            }
        } catch (\Exception $e) {
            return false;
        }

        $cacheKey = $applicationId . ':' . $state;
        $eligibleStates = ['in-process', 'waiting for decision', 'waiting for documents', 'form-completed'];

        return self::$eligibility[$cacheKey] = in_array($state, $eligibleStates, true)
            && self::createdAfterRelease($createdAt, $applicationId);
    }

    /**
     * The release-date comparison.
     *
     * Both failure modes resolve to "out of scope", which is the safe direction:
     * an unknown application age must not licence a decision, and a malformed
     * RELEASE_DATE must not silently open the feature to every historical
     * application. A bad configured date is logged, because the symptom
     * otherwise is a feature that quietly does nothing.
     */
    protected static function createdAfterRelease($createdAt, int $applicationId): bool
    {
        $release = strtotime((string) self::releaseDate());

        if ($release === false) {
            self::log(['title' => 'Redis-Cache', 'message' => 'DatapointHelper: unparseable release date, feature disabled', 'release_date' => self::releaseDate(), 'application_id' => $applicationId]);
            return false;
        }

        if (empty($createdAt)) {
            return false;
        }

        $created = strtotime((string) $createdAt);

        return $created !== false && $created >= $release;
    }

    /**
     * The configured go-live timestamp.
     */
    public static function releaseDate(): string
    {
        return (string) env('DATAPOINT_RELEASE_DATE', self::RELEASE_DATE);
    }

    /**
     * The Redis hash key for an application.
     */
    public static function datapointKey($applicationId): string
    {
        return self::KEY_PREFIX . $applicationId . self::KEY_SUFFIX; // loan:278525:datapoints
    }

    /**
     * The Redis hash key holding an application's dispatch guards.
     */
    public static function jobDispatchKey($applicationId): string
    {
        return self::KEY_PREFIX . $applicationId . self::JOB_KEY_SUFFIX; // loan:278525:dispatch_guard
    }

    /**
     * The dispatch guard field for a tier and applicant type.
     */
    public static function jobTypeKey(int $applicantType, string $tier = 'approval'): string
    {
        $prefix = isset(self::$tiers[$tier]) ? self::$tiers[$tier]['guard'] : self::AUTODECISION_JOB_PREFIX;
        return $prefix . self::normaliseApplicantType($applicantType); // autodecision_dispatched_1
    }

    /**
     * Compute every data point for an application straight from the database.
     * This is the authoritative definition of how each flag is derived — the
     * observers only mirror it for the subset of fields they own.
     *
     * Every flag but one comes from a column. The exception is
     * co_applicant_rejection_executed, which is read from the dispatch guard —
     * see coApplicantRejectionExecuted(). Pass the caller's connection to reuse
     * it; one is opened here if not.
     *
     * @param  int                    $applicationId
     * @param  RawRedisConnector|null $redis Reuse an already-open connection
     * @return array field => 0|1, or [] if the application does not exist
     */
    public static function buildFromDatabase(int $applicationId, RawRedisConnector $redis = null): array
    {
        $application = DB::table('consumer_application')->where('id', $applicationId)->first();

        if (!$application) {
            return [];
        }

        $element = DB::table('consumer_application_element')->where('application_id', $applicationId)->first();

        // A denied consent must not satisfy the tier, hence the status filter.
        //
        // The applicant_type filter matters just as much. consent_log rows carry
        // an applicant_type (default 1, added 2023_11_03), and without the filter
        // a CO-APPLICANT's agreed consent satisfied the APPLICANT's rejection
        // tier — enough on its own to dispatch an applicant decision on an
        // application where the applicant had never consented.
        $consented = array_map('intval', DB::table('consent_log')
            ->where('application_id', $applicationId)
            ->where('status', 1)
            ->whereNull('deleted_at')
            ->pluck('applicant_type')
            ->all());

        $flags = [
            'consent_log' => in_array(1, $consented, true) ? 1 : 0,
            // Synthetic — no column backs it, so it is seeded here and the
            // column loop below skips it.
            'co_applicant_rejection_executed' => self::coApplicantRejectionExecuted($applicationId, $redis),
        ];

        foreach (self::tracked() as $field) {
            if (isset($flags[$field])) {
                continue;
            }

            if (property_exists($application, $field)) {
                $flags[$field] = !empty($application->{$field}) ? 1 : 0;
                continue;
            }

            if ($element && property_exists($element, $field)) {
                $flags[$field] = !empty($element->{$field}) ? 1 : 0;
                continue;
            }

            $flags[$field] = 0;
            if ($element) {
                self::log(['message' => 'DatapointHelper: data point matches no column on consumer_application or consumer_application_element', 'field'=> $field, 'application_id'=>$applicationId]);
            }
        }
        return $flags;
    }

    /**
     * Has the co-applicant rejection tier already been dispatched?
     *
     * co_applicant_rejection_executed is the one data point with no column
     * behind it, so a rebuild has to source it from something that outlives the
     * data point hash. The dispatch guard is exactly that: rejection_dispatched_2
     * on loan:{id}:dispatch_guard, held for JOB_TTL_SECONDS from the moment the
     * co-applicant rejection job reached SQS.
     *
     * Deriving it anywhere else does not work. RunRejectionJob sets the flag
     * through sync() once the co-applicant policy has run, but the flag lives in
     * the short-TTL data point hash — so the next refresh() rebuilt it as 0 while
     * the guard stayed claimed, and nothing was left that could ever set it
     * again. The applicant's approval tier then sat permanently one field short.
     *
     * The flag therefore now means "the co-applicant rejection job has been
     * dispatched", not "it has finished". Within one attemptDispatch() pass that
     * distinction is invisible — the data points are read before any dispatch, so
     * the snapshot approval is judged against still carries 0. Across calls it
     * means a refresh() landing while the co-applicant job is still queued can
     * let the applicant's approval go out first. That is the trade the durable
     * source buys.
     *
     * A Redis failure degrades to 0, which withholds an approval rather than
     * issuing one — the safe direction.
     */
    protected static function coApplicantRejectionExecuted(int $applicationId, RawRedisConnector $redis = null): int
    {
        try {
            $redis = $redis ?: new RawRedisConnector();

            $guards = $redis->hgetall(self::jobDispatchKey($applicationId));

            return empty($guards[self::jobTypeKey(2, 'rejection')]) ? 0 : 1;
        } catch (\Exception $e) {
            self::log(['title' => 'Redis-Cache', 'message' => 'DatapointHelper: co-applicant rejection guard unreadable, treating as not dispatched', 'error' => $e->getMessage(), 'line' => $e->getLine(), 'application_id' => $applicationId]);
            return 0;
        }
    }

    /**
     * The one write path for the data point hash. Every writer goes through
     * here so that change detection lives in a single place.
     *
     * The HGETALL before the write is what makes a notification possible at
     * all. Redis has no "set and give me the previous value" for hash fields,
     * and HMSET/HSET replies count fields *created*, not fields whose value
     * changed — so a 1 -> 0 flip is invisible unless you read first.
     *
     * That same read also detects a cold start with no extra EXISTS: an empty
     * hash cannot exist in Redis, so an empty reply means the key is gone.
     *
     * Costs one extra round trip per save on the observer hot path, on the
     * connection the caller already has open.
     *
     * @param  array $fields field => 0|1; may be a subset of the tracked fields
     * @return array The hash as it now stands (before-snapshot merged with $fields)
     */
    public static function write(RawRedisConnector $redis, int $applicationId, array $fields): array
    {
        // Pre-release applications are not tracked at all — no flags, and so no
        // notification either, since notifyOnChange() is reached only from here.
        if (!self::isEligible($applicationId)) {
            return [];
        }

        $key = self::datapointKey($applicationId);

        $before = $redis->hgetall($key);

        $redis->hmset($key, $fields);
        $redis->expire($key, self::TTL_SECONDS);

        $after = array_merge($before, $fields);

        self::notifyOnChange($applicationId, $before, $after, $fields);

        return $after;
    }

    /**
     * Fire the readiness notification when the hash was cold or a flag flipped.
     *
     * A cold start is delivered as a **direct call** rather than through SQS:
     * the hash was empty, so this is the moment the application enters the
     * tracker and there is nothing to coalesce with — the notification is the
     * baseline every later delta is read against. Subsequent flips go to SQS,
     * because they land on the hot save path where a save must not wait on a
     * notification channel.
     *
     * A cold start is *also* a change in every field it writes, so only one of
     * the two fires: cold start wins and the flips are reported as its changes.
     *
     * Never throws — a notification is not worth failing a loan application
     * save over. Note the corollary: the inline cold-start call runs inside the
     * caller's request, so anything notify() eventually does must not write to
     * an observed model, or the observer will re-enter here.
     */
    protected static function notifyOnChange(int $applicationId, array $before, array $after, array $written): void
    {
        try {
            if (!self::isRedisFlow()) {
                return;
            }

            $coldStart = empty($before);

            // Redis hands back strings ('1'/'0') while observers write ints, so
            // both sides are cast before comparison.
            //
            // A field absent from the before-snapshot is reported as old = null,
            // because "never recorded" and "recorded as missing" are worth
            // telling apart — but it is *compared* as 0, because downstream they
            // are the same fact: pending() reads absent and 0 identically. So a
            // field first appearing in the hash as 0 is not a flip and does not
            // notify, while absent -> 1 does.
            $changes = [];
            foreach ($written as $field => $value) {
                $old = isset($before[$field]) ? (int) $before[$field] : null;
                $new = (int) $value;

                if (($old ?? 0) === $new) {
                    continue;
                }

                $changes[$field] = ['old' => $old, 'new' => $new];
            }

            if (!$coldStart && empty($changes)) {
                return;
            }

            $flags = self::flagsFrom($after);

            // The notification describes the whole application, not one
            // applicant — pending/ready are keyed by applicant type, and the
            // co-applicant key is present only when there is a co-applicant.
            $payload = [
                'id' => $applicationId,
                'trigger' => $coldStart ? self::NOTIFY_COLD_START : self::NOTIFY_CHANGED,
                'has_co_applicant' => self::hasCoApplicant($flags),
                'changes' => $changes,
                'flags' => $flags,
                'pending' => self::pendingByApplicantFrom($flags),
            ];

            if ($coldStart) {
                (new DatapointNotificationJob($payload))->handle();
                return;
            }

            DatapointNotificationJob::dispatch($payload)->onQueue(env('SQS_QUEUE', 'default'));
        } catch (\Exception $e) {
            self::log(['title' => 'Redis-Cache', 'message' => 'DatapointHelper: readiness notification failed', 'error' => $e->getMessage(), 'line' => $e->getLine(), 'trace' => $e->getTraceAsString(), 'application_id' => $applicationId]);
        }
    }

    /**
     * Narrow a raw hash to the tracked data points, as ints.
     *
     * A live hash can carry fields nothing reads any more — flags renamed in a
     * past release, or a legacy in-hash dispatch guard — so the notification
     * reports the tracked set and nothing else.
     *
     * @return array field => 0|1, one entry per tracked data point
     */
    protected static function flagsFrom(array $datapoints): array
    {
        $flags = [];
        foreach (self::tracked() as $field) {
            $flags[$field] = empty($datapoints[$field]) ? 0 : 1;
        }

        return $flags;
    }

    /**
     * The data points a snapshot is still missing, per tier, for one applicant
     * type. Tiers that do not apply to that applicant type are omitted
     * entirely rather than reported as complete.
     *
     * @return array ['approval' => [...missing], 'rejection' => [...missing]]
     */
    protected static function pendingFrom(array $datapoints, int $applicantType = 1): array
    {
        $pending = [];
        foreach (array_keys(self::$tiers) as $tier) {
            $required = self::required($tier, $applicantType, $datapoints);

            if (empty($required)) {
                continue;
            }

            $pending[$tier] = [];
            foreach ($required as $field) {
                if (empty($datapoints[$field])) {
                    $pending[$tier][] = $field;
                }
            }
        }

        return $pending;
    }

    /**
     * The missing data points for every applicant type on the application,
     * keyed by type.
     *
     * Key 2 is present only when the application actually has a co-applicant.
     * Its absence is the signal "no co-applicant", which is a different fact
     * from "co-applicant with six fields outstanding" — reporting the latter on
     * a single-applicant application would have every such application looking
     * permanently incomplete.
     *
     * @return array [1 => ['approval' => [...], 'rejection' => [...]], 2 => ['rejection' => [...]]]
     */
    protected static function pendingByApplicantFrom(array $datapoints): array
    {
        $pending = [];
        foreach (self::applicantTypesFor($datapoints) as $applicantType) {
            $pending[$applicantType] = self::pendingFrom($datapoints, $applicantType);
        }

        return $pending;
    }

    /**
     * The applicant types worth evaluating for an application: always the
     * applicant, plus the co-applicant when one exists.
     *
     * @return array<int>
     */
    protected static function applicantTypesFor(array $datapoints): array
    {
        return self::hasCoApplicant($datapoints) ? [1, 2] : [1];
    }

    /**
     * Force the Redis hash to match the database for every data point.
     *
     * Use this at choke points reached by sources that do not all write through
     * Eloquent (external APIs, query-builder updates, document capture), where
     * the hash may have drifted or never been written at all.
     *
     * Touches only the data point hash — the dispatch guards live in their own
     * key, so no refresh, expiry or DEL of this hash can re-arm a job.
     *
     * Does not dispatch — call dispatchIfReady() once the caller's own
     * eligibility gates have passed. Never throws.
     *
     * @return array The flags written, or [] if nothing was written
     */
    public static function refresh(int $applicationId): array
    {
        try {
            // Checked before buildFromDatabase() so a pre-release application
            // does not pay for three queries write() would then discard.
            if (!self::isEligible($applicationId)) {
                return [];
            }

            // One connection for both the guard read inside buildFromDatabase()
            // and the write that follows.
            $redis = new RawRedisConnector();

            $flags = self::buildFromDatabase($applicationId, $redis);

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

            self::write($redis, $applicationId, $flags);

            return $flags;
        } catch (\Exception $e) {
            self::log(['title' => 'Redis-Cache', 'error'=>$e->getMessage(), 'line' => $e->getLine(), 'trace' => $e->getTraceAsString(), 'application_id'=>$applicationId]);
            return [];
        }
    }

    /**
     * Entry point for writers that do not go through an Eloquent observer and
     * know only their own fields. Records the flags, refreshes the TTL, and
     * (unless told otherwise) evaluates the tier.
     *
     * Prefer refresh() when the caller can afford to re-read everything — it
     * repairs drift instead of only adding to it.
     *
     * Never throws.
     *
     * @param  array $fields        field => 0|1
     * @param  int   $applicantType 1 = applicant, 2 = co-applicant
     * @param  bool  $dispatch      evaluate the tier and dispatch if complete
     * @return bool  true only if a job was dispatched by this call
     */
    public static function sync(int $applicationId, array $fields, int $applicantType = 1, bool $dispatch = true): bool
    {
        if (empty($applicationId) || empty($fields)) {
            return false;
        }

        try {
            $redis = new RawRedisConnector();

            self::write($redis, $applicationId, $fields);

            if (!$dispatch) {
                return false;
            }

            return self::maybeDispatch($applicationId, $redis, self::datapointKey($applicationId), $fields, $applicantType);
        } catch (\Exception $e) {
            self::log(['title' => 'Redis-Cache', 'error'=>$e->getMessage(), 'line' => $e->getLine(), 'trace' => $e->getTraceAsString(), 'application_id'=>$applicationId]);
            return false;
        }
    }

    /**
     * Evaluate every tier for an application whose eligibility the caller has
     * already established, and dispatch each one whose data points are all in.
     *
     * There is no own-field short-circuit here — the caller wrote nothing, so
     * the Redis hash is the only source of truth.
     *
     * Never throws.
     *
     * @param  int|null $applicantType An explicit type narrows evaluation to it;
     *                                 null evaluates the applicant plus the
     *                                 co-applicant when one exists.
     * @return bool     true if at least one job was dispatched by this call
     */
    public static function dispatchIfReady(int $applicationId, int $applicantType = null): bool
    {
        try {
            $redis = new RawRedisConnector();

            return self::attemptDispatch($applicationId, $redis, self::datapointKey($applicationId), self::allCandidates($applicantType)
            );
        } catch (\Exception $e) {
            self::log(['title' => 'Redis-Cache', 'error'=>$e->getMessage(), 'line' => $e->getLine(), 'trace' => $e->getTraceAsString(), 'application_id'=>$applicationId]);
            return false;
        }
    }

    /**
     * The data points still missing, per tier, for ONE applicant type.
     *
     * The flat ['approval' => [...], 'rejection' => [...]] shape is what the
     * admin edit screen consumes, so it is kept for the default applicant —
     * see pendingByApplicant() for the co-applicant too.
     *
     * Reads the hash once. Pass $rebuildIfEmpty to repair an application whose
     * hash was never written or has expired — that is the self-healing path
     * used by the admin edit screen.
     *
     * Never throws.
     *
     * @return array ['rejection' => [...missing], 'approval' => [...missing]]
     */
    public static function pending(int $applicationId, bool $rebuildIfEmpty = false, int $applicantType = 1): array
    {
        // Empty, not "everything pending". A pre-release application has no hash
        // and write() will not create one, so deriving from the empty hash would
        // report all 24 fields as outstanding and the banner would claim a long
        // since decided application is incomplete.
        if (!self::isEligible($applicationId)) {
            return [];
        }

        return self::pendingFrom(self::readDatapoints($applicationId, $rebuildIfEmpty), $applicantType);
    }

    /**
     * The data points still missing for every applicant type on the
     * application, keyed by type. Key 2 appears only when a co-applicant
     * exists.
     *
     * Never throws.
     *
     * @return array [1 => ['approval' => [...], 'rejection' => [...]], 2 => ['rejection' => [...]]]
     */
    public static function pendingByApplicant(int $applicationId, bool $rebuildIfEmpty = false): array
    {
        // See pending() — empty, not "everything pending".
        if (!self::isEligible($applicationId)) {
            return [];
        }

        return self::pendingByApplicantFrom(self::readDatapoints($applicationId, $rebuildIfEmpty));
    }

    /**
     * Read the data point hash, optionally rebuilding it from the database when
     * it has expired or was never written. Never throws — a Redis problem
     * degrades to "everything is pending", not to an exception.
     */
    protected static function readDatapoints(int $applicationId, bool $rebuildIfEmpty): array
    {
        try {
            $redis = new RawRedisConnector();
            $datapoints = $redis->hgetall(self::datapointKey($applicationId));

            if (empty($datapoints) && $rebuildIfEmpty) {
                $datapoints = self::refresh($applicationId);
            }

            return $datapoints;
        } catch (\Exception $e) {
            self::log(['title' => 'Redis-Cache', 'error'=>$e->getMessage(), 'line' => $e->getLine(), 'trace' => $e->getTraceAsString(), 'application_id'=>$applicationId]);
            return [];
        }
    }

    /**
     * After an observer writes its own fields to Redis, call this to check
     * whether any tier is now complete and dispatch its job exactly once.
     *
     * Flow:
     *  1. Short-circuit per (applicant type, tier): a 0 in an own field rules
     *     out only the tiers that require that field — no Redis read if it
     *     rules out all of them.
     *  2. hgetall the guards. Every candidate already claimed → stop here.
     *  3. hgetall the data points, only for the candidates still unclaimed.
     *  4. Check the tier's fields are all present for that applicant type.
     *  5. Claim the guard atomically, then dispatch.
     *
     * @param  int               $id            Application ID
     * @param  RawRedisConnector $redis         Reuse the already-open connection from the observer
     * @param  string            $key           The hash key (loan:{id}:datapoints)
     * @param  array             $ownFields     The fields this observer just wrote (field => 0|1)
     * @param  int|null          $applicantType An explicit type narrows evaluation to it;
     *                                          null evaluates both where a co-applicant exists.
     * @return bool              true only if a job was dispatched by this call
     */
    public static function maybeDispatch(int $id, RawRedisConnector $redis, string $key, array $ownFields, int $applicantType = null): bool
    {
        // Step 1: Short-circuit. Own fields are already in memory from the
        // model, so a half-filled application costs zero Redis reads.
        $candidates = self::candidatesNotBlockedBy($ownFields, $applicantType);

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

        return self::attemptDispatch($id, $redis, $key, $candidates);
    }

    /**
     * Every (applicant type, tier) pair that exists, optionally narrowed to one
     * applicant type.
     *
     * With no applicant type given this includes type 2 unconditionally — the
     * caller has not read the hash yet, so it cannot know whether there is a
     * co-applicant. attemptDispatch() drops type 2 once it has the data points
     * in hand.
     *
     * @return array [applicantType => [tier, ...]]
     */
    protected static function allCandidates(int $applicantType = null): array
    {
        $types = $applicantType === null ? [1, 2] : [self::normaliseApplicantType($applicantType)];

        $candidates = [];
        foreach ($types as $type) {
            foreach (array_keys(self::$tiers) as $tier) {
                if (!empty(self::required($tier, $type))) {
                    $candidates[$type][] = $tier;
                }
            }
        }

        return $candidates;
    }

    /**
     * The (applicant type, tier) pairs that a set of just-written flags does not
     * rule out. A pair survives only if the tier applies to that applicant type
     * and none of the fields it requires was just written as 0.
     *
     * Per-pair rather than per-tier because the two applicant types share a
     * write: ConsumerApplicationObserver stamps all 11 applicant flags and all 6
     * co-applicant flags on one save. A blank offer_id must kill applicant
     * approval without touching co-applicant rejection, and a blank
     * co_applicant_dob must kill co-applicant rejection without touching
     * anything of the applicant's.
     *
     * co_applicant_name gets a second use here. The observer that owns it writes
     * it on every save, so when it appears in $ownFields we know for certain
     * whether a co-applicant exists and can drop the type-2 pairs with zero
     * Redis reads. When it is absent — the element observer, a partial sync() —
     * type 2 stays a candidate and the hash settles it.
     *
     * @param  array $ownFields field => 0|1
     * @return array [applicantType => [tier, ...]]
     */
    protected static function candidatesNotBlockedBy(array $ownFields, int $applicantType = null): array
    {
        $candidates = self::allCandidates($applicantType);

        if ($applicantType === null && array_key_exists('co_applicant_name', $ownFields) && !self::hasCoApplicant($ownFields)) {
            unset($candidates[2]);
        }

        $missing = [];
        foreach ($ownFields as $field => $value) {
            if (empty($value)) {
                $missing[] = $field;
            }
        }

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

        $surviving = [];
        foreach ($candidates as $type => $tiers) {
            foreach ($tiers as $tier) {
                if (!array_intersect($missing, self::required($tier, $type))) {
                    $surviving[$type][] = $tier;
                }
            }
        }

        return $surviving;
    }

    /**
     * Are all the data points a tier needs present for this applicant type?
     *
     * An empty field list means the tier does not apply to that applicant type,
     * which reads as NOT complete. This is the one place that distinction is
     * enforced — see required().
     */
    protected static function tierComplete(array $datapoints, string $tier, int $applicantType): bool
    {
        $required = self::required($tier, $applicantType, $datapoints);

        if (empty($required)) {
            return false;
        }

        foreach ($required as $field) {
            if (empty($datapoints[$field])) {
                return false;
            }
        }

        return true;
    }

    /**
     * Steps 2-5, shared by every entry point.
     *
     * Reads the guards first and the data points only if some tier is still
     * unclaimed, so an application whose jobs have all been sent costs exactly
     * one HGETALL — and one that has not costs two, no matter how many tiers
     * there are or which caller is asking.
     *
     * Applicant types are evaluated in ascending order, tiers within a type in
     * registry order — so the applicant's decision always goes out before the
     * co-applicant's, and approval before rejection.
     *
     * @param  string $key        The data point hash key (loan:{id}:datapoints)
     * @param  array  $candidates [applicantType => [tier, ...]] to consider
     * @return bool   true if at least one job was dispatched by this call
     */
    protected static function attemptDispatch(int $id, RawRedisConnector $redis, string $key, array $candidates): bool
    {
        if (!self::isRedisFlow() || empty($candidates) || !self::isEligible($id)) {
            return false;
        }

        $guardKey = self::jobDispatchKey($id);
        $guards = $redis->hgetall($guardKey); // checking if the job is already dispatched

        krsort($candidates); // co-applicant before applicant

        $open = [];
        foreach ($candidates as $applicantType => $tiers) {
            // Registry order is priority order — see $tiers.
            foreach (array_keys(self::$tiers) as $tier) {
                if (!in_array($tier, $tiers, true)) {
                    continue;
                }
                if (empty($guards[self::jobTypeKey($applicantType, $tier)])) {
                    $open[$applicantType][] = $tier;
                }
            }
        }

        if (empty($open)) {
            return false;
        }

        $datapoints = $redis->hgetall($key); // the data points, now that we know at least one tier can still dispatch.

        // The definitive co-applicant check. Candidate resolution had to guess
        // when co_applicant_name was not among the caller's own fields; now that
        // the hash is in hand, drop the co-applicant entirely if there isn't one.
        // Without this an application with no co-applicant would keep evaluating
        // six fields that will never be filled.
        if (!self::hasCoApplicant($datapoints)) {
            unset($open[2]);

            if (empty($open)) {
                return false;
            }
        }

        // Keyed by applicant type AND tier: subsumption is per applicant, so an
        // approval dispatch for the applicant must not suppress the
        // co-applicant's rejection tier. AutoDecisionJob runs the APPLICANT
        // rejection policy; the co-applicant policy is a different rule set
        // (auto_reject_rule.applicant_type = 2) and nothing has run it.
        $claimed = [];
        $dispatched = false;

        foreach ($open as $applicantType => $tiers) {
            foreach ($tiers as $tier) {
                if (isset($claimed[$applicantType . ':' . $tier])) {
                    continue;
                }

                // Step 4.
                if (!self::tierComplete($datapoints, $tier, $applicantType)) {
                    continue;
                }

                $config = self::$tiers[$tier];
                $guard = self::jobTypeKey($applicantType, $tier);

                // Guards used to live in the data point hash. Honour a legacy one
                // and move it across, so deploying this change cannot re-dispatch
                // a job that was already sent.
                if (!empty($datapoints[$guard])) {
                    self::setJobTypeKey($redis, $guardKey, $guard);
                    continue;
                }

                // Step 5: Claim the guard BEFORE dispatching. HSETNX is atomic, so
                // of two concurrent callers that both see a complete tier only one
                // wins the claim and only one message reaches SQS.
                if (!self::setJobTypeKey($redis, $guardKey, $guard)) {
                    continue;
                }

                try {
                    $job = $config['job'];
                    $job::dispatch(['id' => $id, 'applicant_type' => $applicantType])->onQueue(env('SQS_QUEUE', 'default'));
                } catch (\Exception $e) {
                    // Nothing reached SQS, so release the claim for a later retry.
                    $redis->hdel($guardKey, $guard);
                    throw $e;
                }

                // set the key for the subjobs
                foreach ($config['subjob'] as $job) {
                    self::setJobTypeKey($redis, $guardKey, self::jobTypeKey($applicantType, $job));
                    $claimed[$applicantType . ':' . $job] = true;
                }

                $dispatched = true;
            }
        }

        return $dispatched;
    }

    /**
     * Set the set SQS job key (dispatch guard key set) since it has already been dispatched
     *
     * @return bool true if this call won the claim
     */
    protected static function setJobTypeKey(RawRedisConnector $redis, string $guardKey, string $guard): bool
    {
        $won = (bool) $redis->hsetnx($guardKey, $guard, '1');
        if ($won) {
            $redis->expire($guardKey, self::JOB_TTL_SECONDS);
        }
        return $won;
    }

    protected static function normaliseApplicantType($applicantType): int
    {
        $applicantType = (int) $applicantType;
        return in_array($applicantType, [1, 2], true) ? $applicantType : 1;
    }
}