Redis & Caching

// Vapor 4.122.2, abridged
public protocol Cache {
    func get<T>(_ key: String, as type: T.Type) -> EventLoopFuture<T?> where T: Decodable
    func set<T>(_ key: String, to value: T?) -> EventLoopFuture<Void> where T: Encodable
    func set<T>(_ key: String, to value: T?, expiresIn expirationTime: CacheExpirationTime?) -> EventLoopFuture<Void> where T: Encodable
    func `for`(_ request: Request) -> Self
}

// Cache+Concurrency.swift: async throws overloads, the ones new code calls
// func get<T>(_ key: String, as type: T.Type) async throws -> T? where T: Decodable & Sendable
// func set<T>(_ key: String, to value: T?, expiresIn: CacheExpirationTime?) async throws where T: Encodable & Sendable
// func delete(_ key: String) async throws

import Fluent
import FluentSQLiteDriver
import Redis
import Vapor

/// Which cache driver backs `app.cache` and `req.cache`.
enum CacheDriver: String {
    case memory, fluent, redis
}

func configure(_ app: Application) async throws {
    app.databases.use(.sqlite(.memory), as: .sqlite)
    app.migrations.add(CreateBookshelf())
    app.migrations.add(CacheEntry.migration) // creates "_fluent_cache"; needed only for the fluent driver

    let driver = Environment.get("CACHE_DRIVER").flatMap(CacheDriver.init(rawValue:)) ?? .memory
    switch driver {
    case .memory:
        app.caches.use(.memory)
    case .fluent:
        app.caches.use(.fluent)
    case .redis:
        try configureRedis(app)
        app.caches.use(.redis)
    }
    try await app.autoMigrate()
    try routes(app)
}

import Fluent
import Vapor

struct BookDTO: Content, Equatable {
    let id: UUID
    let title: String
    let priceCents: Int
}

struct PriceUpdate: Content { let priceCents: Int }

func routes(_ app: Application) throws {
    let counter = UpstreamCounter()

    // GET /catalog/:isbn -> cache-aside around a (stubbed) external API
    app.get("catalog", ":isbn") { req async throws -> Response in
        let isbn = try req.parameters.require("isbn")
        let catalog = CachedCatalog(cache: req.cache, ttl: .minutes(10)) { isbn in
            let n = await counter.next()
            return BookMetadata(isbn: isbn, title: "Upstream title #\(n)", pages: 300)
        }
        let (value, hit) = try await catalog.metadata(isbn: isbn)
        let response = Response(status: .ok)
        try response.content.encode(value)
        response.headers.replaceOrAdd(name: "X-Cache", value: hit ? "HIT" : "MISS")
        return response
    }

    // DELETE /catalog/:isbn -> explicit invalidation
    app.delete("catalog", ":isbn") { req async throws -> HTTPStatus in
        let isbn = try req.parameters.require("isbn")
        try await req.cache.delete(CachedCatalog.key(for: isbn))
        return .noContent
    }

    // GET /books/:id -> database row cached as a DTO; PUT invalidates
    app.get("books", ":id") { req async throws -> Response in
        let id = try req.parameters.require("id", as: UUID.self)
        let key = "book:v1:\(id.uuidString.lowercased())"
        let response: Response
        if let cached = try await req.cache.get(key, as: BookDTO.self) {
            response = try await cached.encodeResponse(for: req)
            response.headers.replaceOrAdd(name: "X-Cache", value: "HIT")
        } else {
            guard let book = try await Book.find(id, on: req.db) else { throw Abort(.notFound) }
            let dto = BookDTO(id: id, title: book.title, priceCents: book.priceCents)
            try await req.cache.set(key, to: dto, expiresIn: .minutes(5))
            response = try await dto.encodeResponse(for: req)
            response.headers.replaceOrAdd(name: "X-Cache", value: "MISS")
        }
        return response
    }

    app.put("books", ":id", "price") { req async throws -> HTTPStatus in
        let id = try req.parameters.require("id", as: UUID.self)
        let update = try req.content.decode(PriceUpdate.self)
        guard let book = try await Book.find(id, on: req.db) else { throw Abort(.notFound) }
        book.priceCents = update.priceCents
        try await book.update(on: req.db)
        try await req.cache.delete("book:v1:\(id.uuidString.lowercased())") // invalidate after the write commits
        return .noContent
    }

    // Seeds one book and returns its id (demo only)
    app.post("seed") { req async throws -> String in
        let author = Author(name: "Ada")
        try await author.save(on: req.db)
        let book = Book(title: "Cache Patterns", isbn: "978-0-00-000000-1", publishedYear: 2026, priceCents: 2500,
                        status: .published, authorID: try author.requireID())
        try await book.save(on: req.db)
        return try book.requireID().uuidString
    }
}

$ curl -si localhost:8091/catalog/978-1 | grep -E "X-Cache|title"     # first call: miss, upstream is called
X-Cache: MISS
{"isbn":"978-1","title":"Upstream title #1","pages":300}
$ curl -si localhost:8091/catalog/978-1 | grep -E "X-Cache|title"     # second call: served from the cache
X-Cache: HIT
{"title":"Upstream title #1","isbn":"978-1","pages":300}
$ curl -s -o /dev/null -w "DELETE %{http_code}\n" -X DELETE localhost:8091/catalog/978-1
DELETE 204
$ curl -si localhost:8091/catalog/978-1 | grep -E "X-Cache|title"     # after invalidation: upstream called again (#2)
X-Cache: MISS
{"isbn":"978-1","title":"Upstream title #2","pages":300}
$ ID=$(curl -s -X POST localhost:8091/seed)
$ curl -si localhost:8091/books/$ID | grep -E "X-Cache|price"          # MISS, then HIT
$ curl -s -o /dev/null -w "PUT %{http_code}\n" -X PUT -H 'content-type: application/json' -d '{"priceCents":3000}' localhost:8091/books/$ID/price
PUT 204
$ curl -si localhost:8091/books/$ID | grep -E "X-Cache|price"          # MISS again, price now 3000
X-Cache: MISS
{"priceCents":3000,"id":"...","title":"Cache Patterns"}

import Vapor

struct BookMetadata: Content, Equatable {
    let isbn: String
    let title: String
    let pages: Int
}

/// Cache-aside around any async source of metadata. The source is a closure so that production code passes
/// `CatalogClient.metadata` (see the HTTP client page) and tests pass a counting stub.
struct CachedCatalog {
    let cache: any Cache
    let ttl: CacheExpirationTime
    let fetch: @Sendable (String) async throws -> BookMetadata

    static func key(for isbn: String) -> String { "catalog:v1:isbn:\(isbn)" }

    func metadata(isbn: String) async throws -> (value: BookMetadata, hit: Bool) {
        let key = Self.key(for: isbn)
        if let cached = try await cache.get(key, as: BookMetadata.self) {
            return (cached, true)
        }
        let fresh = try await fetch(isbn)
        try await cache.set(key, to: fresh, expiresIn: ttl)
        return (fresh, false)
    }

    func invalidate(isbn: String) async throws {
        try await cache.delete(Self.key(for: isbn))
    }
}

/// Counts upstream calls so the example can show hits and misses.
actor UpstreamCounter {
    private(set) var calls = 0
    func next() -> Int { calls += 1; return calls }
}

// Package.swift (dependencies)
.package(url: "https://github.com/vapor/redis.git", from: "4.14.0"),

// target dependencies
.product(name: "Redis", package: "redis"),

// RedisExtras.swift, continued

// MARK: Single-flight lock against a cache stampede

enum StampedeGuard {
    /// Returns true for exactly one caller per key and lock lifetime (SET key value NX EX seconds).
    static func tryAcquire(_ req: Request, key: String, seconds: Int) async throws -> Bool {
        let result = try await req.redis.set(
            RedisKey("lock:\(key)"),
            to: req.id,
            onCondition: .keyDoesNotExist,
            expiration: .seconds(seconds)
        ).get()
        return result == .ok
    }

    static func release(_ req: Request, key: String) async throws {
        _ = try await req.redis.delete(RedisKey("lock:\(key)")).get()
    }
}

/// A TTL with up to 20 percent random jitter, so entries written together do not all expire together.
func jitteredTTL(seconds: Int) -> CacheExpirationTime {
    .seconds(seconds + Int.random(in: 0...(seconds / 5)))
}

// RedisExtras.swift, continued

// MARK: Fixed-window rate limiter on INCR + EXPIRE

struct RateLimitMiddleware: AsyncMiddleware {
    let limit: Int
    let windowSeconds: Int

    func respond(to request: Request, chainingTo next: any AsyncResponder) async throws -> Response {
        let client = request.remoteAddress?.ipAddress ?? "unknown"
        let window = Int(Date().timeIntervalSince1970) / windowSeconds
        let key = RedisKey("rl:\(client):\(window)")
        let count = try await request.redis.increment(key).get()
        if count == 1 {
            _ = try await request.redis.expire(key, after: .seconds(Int64(windowSeconds))).get()
        }
        guard count <= limit else {
            throw Abort(.tooManyRequests, headers: ["Retry-After": "\(windowSeconds)"])
        }
        return try await next.respond(to: request)
    }
}

// RedisExtras.swift, continued

// MARK: Pub/Sub

extension RedisChannelName {
    static var bookChanged: RedisChannelName { "bookshelf:book-changed" }
}

struct BookChanged: Codable, Sendable {
    let bookID: UUID
    let instance: String
}

/// Subscribes at start-up; every instance receives every message, including the one that published it.
struct BookChangeSubscriber: LifecycleHandler {
    let instance: String

    func didBoot(_ application: Application) throws {
        let logger = application.logger
        let me = instance
        Task {
            do {
                try await application.redis.subscribe(
                    to: [.bookChanged],
                    messageReceiver: { channel, message in
                        guard let text = message.string,
                              let event = try? JSONDecoder().decode(BookChanged.self, from: Data(text.utf8)) else {
                            logger.warning("Ignoring malformed message on \(channel)")
                            return
                        }
                        logger.info("Book \(event.bookID) changed on \(event.instance); I am \(me)")
                        // Here: evict the in-process cache entry for event.bookID.
                    },
                    onSubscribe: { key, count in logger.info("Subscribed to \(key) (\(count) active)") },
                    onUnsubscribe: { key, count in logger.info("Unsubscribed from \(key) (\(count) left)") }
                )
            } catch {
                logger.error("Redis subscription failed: \(error)")
            }
        }
    }
}

func publishBookChanged(_ req: Request, bookID: UUID, instance: String) async throws {
    let payload = String(decoding: try JSONEncoder().encode(BookChanged(bookID: bookID, instance: instance)), as: UTF8.self)
    let receivers = try await req.redis.publish(payload, to: .bookChanged).get()
    req.logger.debug("Delivered to \(receivers) subscriber(s)")
}

@testable import App
import Testing
import Vapor
import VaporTesting

@Suite("Cached handlers")
struct HandlerTests {
    @Test("the catalog route misses once, then hits, then misses after invalidation")
    func catalogCacheAside() async throws {
        let app = try await Application.make(.testing)
        do {
            try await configure(app) // CACHE_DRIVER is unset, so this is the memory driver
            try await app.testing().test(.GET, "catalog/978-1") { res async in
                #expect(res.headers.first(name: "X-Cache") == "MISS")
                #expect((try? res.content.decode(BookMetadata.self))?.title == "Upstream title #1")
            }.test(.GET, "catalog/978-1") { res async in
                #expect(res.headers.first(name: "X-Cache") == "HIT")
                #expect((try? res.content.decode(BookMetadata.self))?.title == "Upstream title #1")
            }.test(.DELETE, "catalog/978-1") { res async in
                #expect(res.status == .noContent)
            }.test(.GET, "catalog/978-1") { res async in
                #expect(res.headers.first(name: "X-Cache") == "MISS")
                #expect((try? res.content.decode(BookMetadata.self))?.title == "Upstream title #2")
            }
        } catch {
            try await app.asyncShutdown()
            throw error
        }
        try await app.asyncShutdown()
    }
}