feat: port Cache::funnel from Laravel

src/cache/src/Limiters/ConcurrencyLimiter.php

@@ -0,0 +1,99 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Hypervel\Cache\Limiters;
+
+use Hypervel\Contracts\Cache\Lock;
+use Hypervel\Contracts\Cache\LockProvider;
+use Hypervel\Support\Sleep;
+use Hypervel\Support\Str;
+use Throwable;
+
+class ConcurrencyLimiter
+{
+    /**
+     * Precomputed slot names. Built once in the constructor.
+     *
+     * @var list<string>
+     */
+    protected array $slots;
+
+    /**
+     * Create a new concurrency limiter instance.
+     *
+     * @param LockProvider $store the cache store instance
+     * @param string $name the name of the limiter
+     * @param int $maxLocks the allowed number of concurrent locks
+     * @param int $releaseAfter the number of seconds a slot should be maintained
+     */
+    public function __construct(
+        protected LockProvider $store,
+        protected string $name,
+        protected int $maxLocks,
+        protected int $releaseAfter,
+    ) {
+        $this->slots = $maxLocks < 1
+            ? []
+            : array_map(fn (int $i): string => $name . $i, range(1, $maxLocks));
+    }
+
+    /**
+     * Attempt to acquire the lock for the given number of seconds.
+     *
+     * @throws LimiterTimeoutException
+     * @throws Throwable
+     */
+    public function block(int $timeout, ?callable $callback = null, int $sleep = 250): mixed
+    {
+        $starting = time();
+
+        $id = Str::random(20);
+
+        while (! $slot = $this->acquire($id)) {
+            if (time() - $timeout >= $starting) {
+                throw new LimiterTimeoutException;
+            }
+
+            Sleep::usleep($sleep * 1000);
+        }
+
+        if (is_callable($callback)) {
+            try {
+                return tap($callback(), function () use ($slot): void {
+                    $this->release($slot);
+                });
+            } catch (Throwable $exception) {
+                $this->release($slot);
+
+                throw $exception;
+            }
+        }
+
+        return true;
+    }
+
+    /**
+     * Attempt to acquire a slot lock.
+     */
+    protected function acquire(string $id): bool|Lock
+    {
+        foreach ($this->slots as $slotName) {
+            $lock = $this->store->lock($slotName, $this->releaseAfter, $id);
+
+            if ($lock->acquire()) {
+                return $lock;
+            }
+        }
+
+        return false;
+    }
+
+    /**
+     * Release the slot lock.
+     */
+    protected function release(Lock $lock): void
+    {
+        $lock->release();
+    }
+}

src/cache/src/Limiters/ConcurrencyLimiterBuilder.php

@@ -0,0 +1,125 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Hypervel\Cache\Limiters;
+
+use DateInterval;
+use DateTimeInterface;
+use Hypervel\Cache\RedisStore;
+use Hypervel\Cache\Repository;
+use Hypervel\Contracts\Cache\LockProvider;
+use Hypervel\Contracts\Cache\Store;
+use Hypervel\Support\InteractsWithTime;
+
+class ConcurrencyLimiterBuilder
+{
+    use InteractsWithTime;
+
+    /**
+     * The maximum number of entities that can hold the lock at the same time.
+     */
+    public int $maxLocks;
+
+    /**
+     * The number of seconds to maintain the lock until it is automatically released.
+     */
+    public int $releaseAfter = 60;
+
+    /**
+     * The number of seconds to block until a lock is available.
+     */
+    public int $timeout = 3;
+
+    /**
+     * The number of milliseconds to wait between attempts to acquire the lock.
+     */
+    public int $sleep = 250;
+
+    /**
+     * Create a new builder instance.
+     *
+     * @param Repository $connection the cache repository instance
+     * @param string $name the name of the limiter
+     */
+    public function __construct(
+        public Repository $connection,
+        public string $name,
+    ) {
+    }
+
+    /**
+     * Set the maximum number of locks that can be obtained per time window.
+     */
+    public function limit(int $maxLocks): static
+    {
+        $this->maxLocks = $maxLocks;
+
+        return $this;
+    }
+
+    /**
+     * Set the number of seconds until the lock will be released.
+     */
+    public function releaseAfter(DateInterval|DateTimeInterface|int $releaseAfter): static
+    {
+        $this->releaseAfter = $this->secondsUntil($releaseAfter);
+
+        return $this;
+    }
+
+    /**
+     * Set the number of seconds to block until a lock is available.
+     */
+    public function block(int $timeout): static
+    {
+        $this->timeout = $timeout;
+
+        return $this;
+    }
+
+    /**
+     * The number of milliseconds to wait between lock acquisition attempts.
+     */
+    public function sleep(int $sleep): static
+    {
+        $this->sleep = $sleep;
+
+        return $this;
+    }
+
+    /**
+     * Execute the given callback if a lock is obtained, otherwise call the failure callback.
+     *
+     * @throws LimiterTimeoutException
+     */
+    public function then(callable $callback, ?callable $failure = null): mixed
+    {
+        try {
+            return $this->createLimiter()->block($this->timeout, $callback, $this->sleep);
+        } catch (LimiterTimeoutException $e) {
+            if ($failure !== null) {
+                return $failure($e);
+            }
+
+            throw $e;
+        }
+    }
+
+    /**
+     * Create the concurrency limiter instance.
+     */
+    protected function createLimiter(): ConcurrencyLimiter
+    {
+        // Type is guaranteed by the LockProvider check in Repository::funnel(),
+        // but Repository::getStore() returns Store so phpstan needs help narrowing.
+        /** @var LockProvider&Store $store */
+        $store = $this->connection->getStore();
+
+        if ($store instanceof RedisStore) {
+            return new RedisConcurrencyLimiter($store, $this->name, $this->maxLocks, $this->releaseAfter);
+        }
+
+        return new ConcurrencyLimiter($store, $this->name, $this->maxLocks, $this->releaseAfter);
+    }
+}

src/cache/src/Limiters/LimiterTimeoutException.php

@@ -0,0 +1,11 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Hypervel\Cache\Limiters;
+
+use Exception;
+
+class LimiterTimeoutException extends Exception
+{
+}

src/cache/src/Limiters/RedisConcurrencyLimiter.php

@@ -0,0 +1,87 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Hypervel\Cache\Limiters;
+
+use Hypervel\Cache\RedisStore;
+use Hypervel\Contracts\Cache\Lock;
+use Hypervel\Redis\LuaScripts;
+use Hypervel\Redis\RedisConnection;
+
+/**
+ * Redis-optimized concurrency limiter.
+ */
+class RedisConcurrencyLimiter extends ConcurrencyLimiter
+{
+    /**
+     * Precomputed prefixed slot keys for Lua KEYS. Built once in the constructor.
+     *
+     * @var list<string>
+     */
+    protected array $prefixedSlots;
+
+    /**
+     * Create a new Redis-optimized concurrency limiter instance.
+     */
+    public function __construct(
+        RedisStore $store,
+        string $name,
+        int $maxLocks,
+        int $releaseAfter,
+    ) {
+        parent::__construct($store, $name, $maxLocks, $releaseAfter);
+
+        $prefix = $store->getPrefix();
+        $this->prefixedSlots = array_map(
+            fn (string $slot): string => $prefix . $slot,
+            $this->slots,
+        );
+    }
+
+    /**
+     * Atomically claim a free slot via a single Lua script.
+     *
+     * Two correctness invariants:
+     *
+     * 1. The Lua script writes the prefixed slot key to Redis and returns the
+     *    UNPREFIXED slot name (e.g. "my-funnel1") so RedisStore::restoreLock()
+     *    prepends the prefix exactly once when constructing the Lock object.
+     *
+     * 2. The owner ID must be pre-packed via $connection->pack() before being
+     *    passed into Lua. phpredis does NOT auto-serialize eval() ARGV (regular
+     *    commands like set() do). RedisLock::release() later packs $this->owner
+     *    before its owner-check Lua, so the value Redis stores at acquire time
+     *    must already be in packed form. If we passed raw $id here, Redis would
+     *    store the raw string, but release would compare against a packed value
+     *    — silent mismatch, slot leaks until TTL. We pass the RAW $id to
+     *    restoreLock() so the returned RedisLock's owner field is raw; release
+     *    will pack it consistently with what we stored.
+     *
+     * Using withConnection() also keeps both pack() and eval() on the same
+     * checked-out pool connection, avoiding two pool roundtrips per attempt.
+     */
+    protected function acquire(string $id): bool|Lock
+    {
+        // Without slots there's nothing to claim. Calling eval with zero KEYS
+        // would error inside Lua via unpack({}) → redis.call('mget') with no args.
+        if ($this->prefixedSlots === []) {
+            return false;
+        }
+
+        /** @var RedisStore $store */
+        $store = $this->store;
+
+        return $store->lockConnection()->withConnection(function (RedisConnection $connection) use ($id, $store): bool|Lock {
+            $packedOwner = $connection->pack([$id])[0];
+
+            $result = $connection->eval(...array_merge(
+                [LuaScripts::acquireConcurrencySlot(), count($this->prefixedSlots)],
+                $this->prefixedSlots,
+                [$this->name, $this->releaseAfter, $packedOwner],
+            ));
+
+            return is_string($result) ? $store->restoreLock($result, $id) : false;
+        });
+    }
+}

src/cache/src/RedisLock.php

@@ -5,6 +5,7 @@
 namespace Hypervel\Cache;
 
 use Hypervel\Contracts\Cache\RefreshableLock;
+use Hypervel\Redis\LuaScripts;
 use Hypervel\Redis\RedisConnection;
 use Hypervel\Redis\RedisProxy;
 use InvalidArgumentException;