<?php

namespace App\Services\Link;

use App\Models\LinkWallet;
use App\Models\ShortLink;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;

use Throwable;

/**
 * Buffers click counter increments in Redis instead of writing them on the
 * request path.
 *
 * Every visit used to run three UPDATEs against the same short_links row plus one
 * against the owner's link_wallets row, all inside one transaction. A popular link
 * therefore serialised its entire traffic through a single row lock, and because
 * two rows were locked in one transaction, concurrent clicks on different links of
 * the same owner could acquire them in opposite orders — which is exactly the
 * "Deadlock found when trying to get lock" that showed up 146 times on one row.
 *
 * Redis INCR has no such contention, so the request path becomes lock-free and the
 * database sees one batched UPDATE per link per flush instead of four per click.
 *
 * Redis is treated as a buffer, never as the source of truth: link_clicks still
 * records every click as an insert-only row, so a lost buffer costs at most a short
 * drift in the denormalised totals, which `link:flush-click-counters --recount`
 * repairs from link_clicks.
 */
class LinkClickCounterBuffer
{
    private const CLICKS = 'link:counters:clicks';

    private const UNIQUE = 'link:counters:unique';

    private const LAST_CLICKED = 'link:counters:last_clicked';

    private const WALLET = 'link:counters:wallet';

    public function __construct(private readonly ClickCounterStore $store) {}

    public function add(ShortLink $link, bool $isUnique): void
    {
        try {
            $this->store->increment(self::CLICKS, (string) $link->id);
            if ($isUnique) {
                $this->store->increment(self::UNIQUE, (string) $link->id);
            }
            $this->store->put(self::LAST_CLICKED, (string) $link->id, now()->toDateTimeString());
            $this->store->increment(self::WALLET, (string) $link->user_id);
        } catch (Throwable $e) {
            // Redis down: fall back to the direct writes rather than lose the count.
            // This reintroduces the contention, but only while Redis is unavailable.
            Log::warning('link click counter buffer unavailable, writing through', [
                'short_link_id' => $link->id,
                'error' => $e->getMessage(),
            ]);

            $this->writeThrough($link, $isUnique);
        }
    }

    /**
     * Drain the buffer into the database.
     *
     * Each hash is read and deleted in one transaction so a concurrent click either
     * lands in the batch being flushed or in the next one, never in neither.
     *
     * @return array{links: int, wallets: int, clicks: int}
     */
    public function flush(): array
    {
        [$clicks, $unique, $lastClicked, $wallets] = $this->drain();

        $totalClicks = 0;
        $linkIds = array_unique([...array_keys($clicks), ...array_keys($unique), ...array_keys($lastClicked)]);

        foreach ($linkIds as $linkId) {
            $delta = (int) ($clicks[$linkId] ?? 0);
            $uniqueDelta = (int) ($unique[$linkId] ?? 0);
            $stamp = $lastClicked[$linkId] ?? null;
            $totalClicks += $delta;

            // One UPDATE per link per flush, replacing three per click.
            $sets = [];
            if ($delta !== 0) {
                $sets['clicks_total'] = DB::raw('clicks_total + '.$delta);
            }
            if ($uniqueDelta !== 0) {
                $sets['unique_clicks_total'] = DB::raw('unique_clicks_total + '.$uniqueDelta);
            }
            if ($stamp !== null) {
                $sets['last_clicked_at'] = $stamp;
            }
            if ($sets !== []) {
                ShortLink::query()->whereKey($linkId)->update($sets);
            }
        }

        foreach ($wallets as $userId => $delta) {
            if ((int) $delta !== 0) {
                LinkWallet::query()->where('user_id', $userId)->increment('clicks_total', (int) $delta);
            }
        }

        return [
            'links' => count($linkIds),
            'wallets' => count($wallets),
            'clicks' => $totalClicks,
        ];
    }

    /**
     * Rebuild the denormalised totals from link_clicks, which is insert-only and
     * therefore authoritative. Use after a Redis loss, or to verify drift.
     */
    public function recount(): int
    {
        $rows = DB::table('link_clicks')
            ->select('short_link_id')
            ->selectRaw('COUNT(*) as total')
            ->selectRaw('SUM(CASE WHEN is_unique = 1 THEN 1 ELSE 0 END) as uniques')
            ->selectRaw('MAX(clicked_at) as last_clicked')
            ->groupBy('short_link_id')
            ->get();

        foreach ($rows as $row) {
            ShortLink::query()->whereKey($row->short_link_id)->update([
                'clicks_total' => (int) $row->total,
                'unique_clicks_total' => (int) $row->uniques,
                'last_clicked_at' => $row->last_clicked,
            ]);
        }

        return $rows->count();
    }

    /**
     * Read-and-clear each hash atomically.
     *
     * @return array{0: array<string,int>, 1: array<string,int>, 2: array<string,string>, 3: array<string,int>}
     */
    private function drain(): array
    {
        $out = [];

        foreach ([self::CLICKS, self::UNIQUE, self::LAST_CLICKED, self::WALLET] as $key) {
            try {
                $out[] = $this->store->drain($key);
            } catch (Throwable $e) {
                Log::warning('link click counter flush could not drain', [
                    'key' => $key,
                    'error' => $e->getMessage(),
                ]);
                $out[] = [];
            }
        }

        return $out;
    }

    private function writeThrough(ShortLink $link, bool $isUnique): void
    {
        try {
            // Single statement, so the row is locked once rather than three times.
            $sets = [
                'clicks_total' => DB::raw('clicks_total + 1'),
                'last_clicked_at' => now(),
            ];
            if ($isUnique) {
                $sets['unique_clicks_total'] = DB::raw('unique_clicks_total + 1');
            }
            ShortLink::query()->whereKey($link->id)->update($sets);

            LinkWallet::query()->where('user_id', $link->user_id)->increment('clicks_total');
        } catch (Throwable $e) {
            // Never let a counter failure break the redirect the visitor came for.
            Log::warning('link click counter write-through failed', [
                'short_link_id' => $link->id,
                'error' => $e->getMessage(),
            ]);
        }
    }
}