import { Json } from 'ox'

import * as Timeout from './Timeout.js'

/** Key-to-value map used by typed stores. */
export type StoreItemMap = Record<string, unknown>

/** Result of an atomic store update. */
export type Change<value, result> =
  | {
      /** Leaves the value unchanged. */
      op: 'noop'
      /** Value returned by the update. */
      result: result
    }
  | {
      /** Replaces the stored value. */
      op: 'set'
      /** Value returned by the update. */
      result: result
      /** New stored value. */
      value: value
    }
  | {
      /** Deletes the stored value. */
      op: 'delete'
      /** Value returned by the update. */
      result: result
    }

/** Typed store with atomic read-modify-write support. */
export type AtomicStore<itemMap extends StoreItemMap = StoreItemMap> = {
  /** Deletes a key. */
  delete<key extends keyof itemMap & string>(key: key): Promise<void>
  /** Gets a typed value. */
  get<key extends keyof itemMap & string>(key: key): Promise<itemMap[key] | null>
  /** Writes a typed value. */
  put<key extends keyof itemMap & string>(key: key, value: itemMap[key]): Promise<void>
  /** Atomically updates a typed value. */
  update<key extends keyof itemMap & string, result>(
    key: key,
    fn: (current: itemMap[key] | null) => Change<itemMap[key], result>,
  ): Promise<result>
}

/** Minimal string store contract used by Tempo API internals. */
export type Store = {
  /** Deletes a key. */
  delete(key: string): Promise<void>
  /** Deletes a key only when its value matches `expected`. */
  deleteIf?: ((key: string, expected: null | string) => Promise<boolean>) | undefined
  /** Gets a string value. Returns `null` for missing or expired keys. */
  get(key: string): Promise<null | string>
  /**
   * Optional native counter: increments an integer value and returns the new
   * count. A missing or expired key restarts at `1`; `ttl` applies like
   * {@link Store.put}. Implement when the backend can execute the
   * read-modify-write in one place (in-memory, inside a Durable Object);
   * consumers go through the {@link increment} helper, which falls back to a
   * get + put when this is absent.
   */
  increment?: ((key: string, options?: Store.PutOptions) => Promise<number>) | undefined
  /** Lists keys. */
  list(options?: {
    /** Restrict results to keys with this prefix. */
    prefix?: string | undefined
  }): Promise<{
    /** Matching keys. */
    keys: readonly {
      /** Key name. */
      name: string
    }[]
  }>
  /** Writes a string value, optionally with a time-to-live. */
  put(key: string, value: string, options?: Store.PutOptions): Promise<void>
  /**
   * Optional atomic compare-and-swap: writes `next` only when the current
   * value equals `expected` (`null` = only-if-absent) and reports whether the
   * write happened. Implement when the backend can execute the compare and
   * write in one place (in-memory, inside a Durable Object); consumers go
   * through the {@link update} helper, which falls back to a get + put when
   * this is absent.
   */
  swap?:
    | ((
        key: string,
        expected: null | string,
        next: string,
        options?: Store.PutOptions,
      ) => Promise<boolean>)
    | undefined
  /**
   * Store kind discriminant. `'state'` marks an authoritative, enumerable
   * store ({@link State}); `'cache'` marks a cache (e.g. the Web Cache,
   * which cannot enumerate keys and silently drops entries).
   */
  type: Store.Type
}

export namespace Store {
  /** Store kind: `'state'` (authoritative, enumerable) or `'cache'` (a cache). */
  export type Type = 'state' | 'cache'

  /** Options for writing a value to a store. */
  export type PutOptions = {
    /**
     * Time-to-live in milliseconds. After this duration the key is treated as
     * absent. Production backends should map this to native expiration
     * (Redis `PEXPIRE`, Cloudflare `expirationTtl`, etc.).
     */
    ttl?: number | undefined
  }
}

/**
 * An authoritative, enumerable {@link Store} for stateful features (webhook
 * subscriptions, rate-limit counters, …). Discriminated by `type: 'state'` so
 * stateful features can require it at the type level and reject cache adapters
 * like the Web Cache ({@link cache}, `type: 'cache'`), which cannot enumerate
 * keys and silently drops entries on eviction. `memory`, `cloudflareKv`, and
 * `durableObject` produce it.
 */
export type State = Store & { type: 'state' }

/** Authoritative string state that also supports typed atomic operations. */
export type AtomicState<itemMap extends StoreItemMap = StoreItemMap> = AtomicStore<itemMap> & State

/**
 * Wraps store operations into a {@link Store}, defaulting to an authoritative
 * {@link State} store. Omit `type` for the common authoritative case; pass
 * `type: 'cache'` to opt into a cache-kind store (non-enumerable, lossy — e.g.
 * the Web Cache).
 */
export function from<const store extends from.Input>(store: store): from.Output<store> {
  // Default the discriminant so only cache-kind stores have to opt in.
  return { ...store, type: store.type ?? 'state' } as never
}

export declare namespace from {
  /** Store operations with an optional kind discriminant (defaults to `'state'`). */
  type Input = Omit<Store, 'type'> & {
    /** Store kind; defaults to `'state'`. Pass `'cache'` for a cache-kind store. */
    type?: Store.Type | undefined
  }

  /** Resolved store kind: {@link State} unless `type: 'cache'` is provided. */
  type Output<store extends Input> = store extends { type: 'cache' } ? Store : State
}

/**
 * Increments an integer counter on a store and returns the new value. A
 * missing or expired key restarts at `1`; `ttl` applies like {@link Store.put}.
 *
 * Uses the store's native {@link Store.increment} when implemented (atomic
 * where the backend executes it in one place — in-memory, inside a Durable
 * Object). Otherwise falls back to a get + put read-modify-write, which
 * costs two round trips on remote stores and can race under concurrency —
 * fine for dev/in-process backends, not for enforced distributed counters.
 */
export async function increment(
  store: Store,
  key: string,
  options: Store.PutOptions = {},
): Promise<number> {
  if (store.increment) return store.increment(key, options)
  const count = Number((await store.get(key)) ?? 0) + 1
  await store.put(key, String(count), options)
  return count
}

/**
 * Atomically transforms a value on a store: reads the current value, applies
 * `fn`, and writes the result, returning the written value.
 *
 * Uses the store's native {@link Store.swap} compare-and-swap when implemented,
 * re-running `fn` on contention (optimistic concurrency, bounded by
 * `attempts`). Otherwise falls back to a get + put read-modify-write, which
 * can lose concurrent writes — fine for dev/in-process backends, not for
 * contended distributed state.
 */
export async function update(
  store: Store,
  key: string,
  fn: (current: null | string) => string,
  options: update.Options = {},
): Promise<string> {
  const { attempts = updateAttempts, ...put } = options

  if (!store.swap) {
    const next = fn(await store.get(key))
    await store.put(key, next, put)
    return next
  }

  for (let attempt = 0; attempt < attempts; attempt++) {
    const current = await store.get(key)
    const next = fn(current)
    if (await store.swap(key, current, next, put)) return next
  }
  throw new UpdateContentionError(key, attempts)
}

const updateAttempts = 8

export declare namespace update {
  /** Options for {@link update}. */
  type Options = Store.PutOptions & {
    /**
     * Maximum compare-and-swap attempts before failing with
     * {@link UpdateContentionError}.
     * @default 8
     */
    attempts?: number | undefined
  }
}

/** Atomically applies a typed state transition without a get-and-put fallback. */
export async function change<result>(
  state: State,
  key: string,
  fn: (current: null | string) => Change<string, result>,
  options: change.Options = {},
): Promise<result> {
  const { attempts = updateAttempts, ...put } = options
  const swap = state.swap
  if (!swap) throw new AtomicStateRequiredError()

  for (let attempt = 0; attempt < attempts; attempt++) {
    const current = await state.get(key)
    const next = fn(current)
    if (next.op === 'noop') return next.result
    if (next.op === 'delete') {
      if (!state.deleteIf) throw new AtomicDeleteRequiredError()
      if (await state.deleteIf(key, current)) return next.result
      continue
    }
    if (await swap(key, current, next.value, put)) return next.result
  }
  throw new UpdateContentionError(key, attempts)
}

export declare namespace change {
  /** Options for one atomic state transition. */
  type Options = update.Options
}

class AtomicStateRequiredError extends Error {
  override name = 'Store.AtomicStateRequiredError'

  constructor() {
    super('Atomic state requires compare-and-swap support.')
  }
}

class AtomicDeleteRequiredError extends Error {
  override name = 'Store.AtomicDeleteRequiredError'

  constructor() {
    super('Atomic deletion requires compare-and-delete support.')
  }
}

/** Creates an in-memory store with TTL support. */
export function memory(options: memory.Options = {}): State {
  // Track value + optional expiration in one cell so `get` can lazily evict
  // expired entries without a background sweeper.
  type Cell = { expiresAt: number | undefined; value: string }
  const cells = new Map<string, Cell>(
    (options.entries ?? []).map(([key, value]) => [key, { expiresAt: undefined, value }]),
  )

  function read(key: string) {
    const cell = cells.get(key)
    if (!cell) return null
    if (cell.expiresAt !== undefined && cell.expiresAt <= Date.now()) {
      cells.delete(key)
      return null
    }
    return cell
  }

  return from({
    async delete(key) {
      cells.delete(key)
    },
    async deleteIf(key, expected) {
      if ((read(key)?.value ?? null) !== expected) return false
      cells.delete(key)
      return true
    },
    async get(key) {
      return read(key)?.value ?? null
    },
    // Native increment: the read-modify-write is synchronous, so concurrent
    // consumers cannot interleave between the read and the write.
    async increment(key, options = {}) {
      const count = Number(read(key)?.value ?? 0) + 1
      const expiresAt = options.ttl === undefined ? undefined : Date.now() + options.ttl
      cells.set(key, { expiresAt, value: String(count) })
      return count
    },
    async list(options = {}) {
      const keys = Array.from(cells.keys())
        .filter((key) => (!options.prefix || key.startsWith(options.prefix)) && read(key))
        .sort()
        .map((name) => ({ name }))

      return { keys }
    },
    async put(key, value, options = {}) {
      const expiresAt = options.ttl === undefined ? undefined : Date.now() + options.ttl
      cells.set(key, { expiresAt, value })
    },
    // Native swap: the compare and write are synchronous, so concurrent
    // consumers cannot interleave between them.
    async swap(key, expected, next, options = {}) {
      if ((read(key)?.value ?? null) !== expected) return false
      const expiresAt = options.ttl === undefined ? undefined : Date.now() + options.ttl
      cells.set(key, { expiresAt, value: next })
      return true
    },
  })
}

export declare namespace memory {
  /** Options for creating an in-memory store. */
  type Options = {
    /** Initial entries. */
    entries?: readonly (readonly [string, string])[] | undefined
  }
}

/**
 * Creates a store backed by a Web Cache API instance (e.g. Cloudflare's
 * colo-local `caches.default`). Values are stored as cache entries keyed by a
 * synthetic URL, with `ttl` mapped to `Cache-Control: max-age`. The Web Cache
 * cannot enumerate keys, so `list` returns no keys — use this for caching
 * (get/put/delete) only, e.g. the origin read cache and edge response cache.
 */
export function cache(webCache: cache.Cache): Store {
  const request = (key: string) => new Request(`https://tempo-api.cache/${encodeURIComponent(key)}`)
  return from({
    async delete(key) {
      await webCache.delete(request(key))
    },
    async get(key) {
      const hit = await webCache.match(request(key))
      return hit ? hit.text() : null
    },
    async list() {
      return { keys: [] }
    },
    async put(key, value, options = {}) {
      // The Web Cache only persists responses with a positive freshness lifetime.
      const seconds = Math.max(1, Math.ceil((options.ttl ?? 0) / 1_000))
      const headers = new Headers({ 'Cache-Control': `max-age=${seconds}` })
      await webCache.put(request(key), new Response(value, { headers }))
    },
    type: 'cache',
  })
}

export declare namespace cache {
  /** Minimal Web Cache API surface used by this adapter. */
  type Cache = {
    /** Deletes a cache entry. */
    delete(request: Request | string): Promise<boolean>
    /** Looks up a cache entry. */
    match(request: Request | string): Promise<Response | undefined>
    /** Stores a cache entry. */
    put(request: Request | string, response: Response): Promise<void>
  }
}

/** Creates a store backed by a Cloudflare Workers KV namespace. */
export function cloudflareKv(namespace: cloudflareKv.Namespace): State {
  return from({
    async delete(key) {
      await namespace.delete(key)
    },
    async get(key) {
      return namespace.get(key)
    },
    async list(options = {}) {
      const keys: { name: string }[] = []
      let cursor: string | undefined

      for (;;) {
        const result = await namespace.list(
          options.prefix === undefined && cursor === undefined
            ? undefined
            : {
                ...(options.prefix === undefined ? {} : { prefix: options.prefix }),
                ...(cursor === undefined ? {} : { cursor }),
              },
        )
        keys.push(...result.keys.map(({ name }) => ({ name })))
        if (result.list_complete) return { keys }
        if (result.cursor === undefined) throw new Error('Cloudflare KV list is missing a cursor.')
        cursor = result.cursor
      }
    },
    async put(key, value, options = {}) {
      const expirationTtl = cloudflareKvExpirationTtl(options.ttl)
      if (expirationTtl === 0) {
        await namespace.delete(key)
        return
      }

      await namespace.put(key, value, expirationTtl === undefined ? undefined : { expirationTtl })
    },
  })
}

export declare namespace cloudflareKv {
  /** Minimal Cloudflare Workers KV namespace shape used by this adapter. */
  type Namespace = {
    /** Deletes a key. */
    delete(key: string): Promise<void>
    /** Gets a text value. */
    get(key: string): Promise<null | string>
    /** Lists keys. */
    list(options?: {
      /** Restrict results to keys with this prefix. */
      prefix?: string | undefined
      /** Continue from a previous page. */
      cursor?: string | undefined
    }): Promise<{
      /** Matching keys. */
      keys: readonly {
        /** Key name. */
        name: string
      }[]
      /** Whether every matching key was returned. */
      list_complete: boolean
      /** Cursor for the next page. */
      cursor?: string | undefined
    }>
    /** Writes a text value. */
    put(
      key: string,
      value: string,
      options?: {
        /** Native Cloudflare KV TTL in seconds. */
        expirationTtl?: number | undefined
      },
    ): Promise<void>
  }
}

/** Creates typed atomic state over namespace RPC, or raw object storage state. */
export function durableObject(namespace: durableObject.Namespace): durableObject.ReturnValue
export function durableObject(
  namespace: durableObject.Namespace,
  options: durableObject.Options,
): durableObject.ReturnValue
export function durableObject<const itemMap extends StoreItemMap>(
  namespace: durableObject.Namespace,
): AtomicState<itemMap>
export function durableObject<const itemMap extends StoreItemMap>(
  namespace: durableObject.Namespace,
  options: durableObject.Options,
): AtomicState<itemMap>
export function durableObject(storage: durableObject.Storage): State
export function durableObject(
  target: durableObject.Namespace | durableObject.Storage,
  options: durableObject.Options = {},
): durableObject.ReturnValue | State {
  if (durableObjectNamespace(target))
    return durableObjectStub(target, options.name ?? ((key) => key))
  return durableObjectStorage(target)
}

function durableObjectStorage(storage: durableObject.Storage): State {
  return from({
    async delete(key) {
      await storage.delete(key)
    },
    // Storage input gates prevent writes between comparison and deletion.
    async deleteIf(key, expected) {
      const current = (await durableObjectRead(storage, key, await storage.get(key)))?.value ?? null
      if (current !== expected) return false
      await storage.delete(key)
      return true
    },
    async get(key) {
      return (await durableObjectRead(storage, key, await storage.get(key)))?.value ?? null
    },
    async list(options = {}) {
      const values = await storage.list<unknown>(options)
      const keys: { name: string }[] = []

      for (const [key, value] of values) {
        if (await durableObjectRead(storage, key, value)) keys.push({ name: key })
      }

      return { keys }
    },
    async put(key, value, options = {}) {
      if (options.ttl !== undefined && options.ttl <= 0) {
        await storage.delete(key)
        return
      }

      await storage.put(key, durableObjectValue(value, options))
    },
    // Native swap: the object is single-threaded and its input gate holds
    // other events while storage operations are in flight, so the compare and
    // write cannot interleave with another request.
    async swap(key, expected, next, options = {}) {
      const current = (await durableObjectRead(storage, key, await storage.get(key)))?.value ?? null
      if (current !== expected) return false
      if (options.ttl !== undefined && options.ttl <= 0) {
        await storage.delete(key)
        return true
      }
      await storage.put(key, durableObjectValue(next, options))
      return true
    },
  })
}

export declare namespace durableObject {
  /** Durable Object state assignable to any typed item map. */
  type ReturnValue = State & {
    /** Gets a contextually typed value. */
    get<value = unknown>(key: string): Promise<value | null>
    /** Writes a typed value. */
    put<value>(key: string, value: value, options?: Store.PutOptions): Promise<void>
    /** Atomically updates a contextually typed value. */
    update<value, result>(
      key: string,
      fn: (current: value | null) => Change<value, result>,
    ): Promise<result>
  }

  /** Minimal Durable Object namespace shape used by this adapter. */
  type Namespace = {
    /** Gets the Durable Object stub for a stable object name. */
    getByName(name: string): Stub
  }

  /** Options for creating a store from a Durable Object namespace. */
  type Options = {
    /** Static pins all keys; a resolver shards them; omission shards by key. */
    name?: string | ((key: string) => string) | undefined
  }

  /** Minimal Cloudflare Durable Object storage shape used by this adapter. */
  type Storage = {
    /** Deletes a key. */
    delete(key: string): Promise<unknown>
    /** Gets a stored value. */
    get<value = unknown>(key: string): Promise<undefined | value>
    /** Lists stored values. */
    list<value = unknown>(options?: {
      /** Restrict results to keys with this prefix. */
      prefix?: string | undefined
    }): Promise<Map<string, value>>
    /** Writes a stored value. */
    put<value>(key: string, value: value): Promise<void>
  }

  /** RPC store operations used by the namespace adapter. */
  type Stub = Omit<Store, 'deleteIf' | 'increment' | 'swap' | 'type'> & {
    /** Atomically deletes a matching value inside the object. */
    deleteIf(key: string, expected: null | string): Promise<boolean>
    /** Atomically increments an integer counter inside the object. */
    increment(key: string, options?: Store.PutOptions): Promise<number>
    /** Atomically compares-and-swaps a value inside the object. */
    swap(
      key: string,
      expected: null | string,
      next: string,
      options?: Store.PutOptions,
    ): Promise<boolean>
  }
}

/** Associates request cache directives with a store used by request handlers. */
export function withRequest(
  store: Store,
  request: withRequest.Request,
  options: withRequest.Options = {},
): Store {
  const source = requestStores.get(store)?.store ?? store
  const scoped = {
    delete: source.delete.bind(source),
    ...(source.deleteIf ? { deleteIf: source.deleteIf.bind(source) } : {}),
    get: source.get.bind(source),
    ...(source.increment ? { increment: source.increment.bind(source) } : {}),
    list: source.list.bind(source),
    put: source.put.bind(source),
    ...(source.swap ? { swap: source.swap.bind(source) } : {}),
    type: source.type,
  }
  requestStores.set(scoped, { request, store: source, waitUntil: options.waitUntil })
  return scoped
}

export declare namespace withRequest {
  /** Request-scoped cache behavior. */
  export type Options = {
    /** Keeps asynchronous cache persistence alive after the response returns. */
    waitUntil?: ((promise: Promise<unknown>) => void) | undefined
  }
  /** Request surface needed to read cache directives. */
  export type Request = {
    /** Reads one request header. */
    header(name: string): string | undefined
  }
}

/**
 * Returns a cached value when present. Otherwise, runs `fetch` and stores its
 * result unless undefined or rejected by `shouldCache`. Cache persistence is
 * best-effort and never delays the fetched value.
 */
export async function memoize<value>(
  fetch: (signal?: AbortSignal) => Promise<value>,
  options: memoize.Options<value>,
): Promise<value> {
  const scoped = requestStores.get(options.store)
  const { key, ttl } = options
  const flightTimeout = options.flightTimeout ?? Timeout.duration
  const request = scoped?.request
  const store = scoped?.store ?? options.store
  const waitUntil = scoped?.waitUntil

  // Request cache directives follow the same semantics as the response cache:
  // no-cache refreshes the entry, while no-store bypasses it entirely.
  const cacheControl = request?.header('cache-control')
  const directives = (cacheControl ?? request?.header('pragma') ?? '').toLowerCase()
  const noStore = directives.includes('no-store')
  const noCache = noStore || directives.includes('no-cache')
  if (noStore)
    return Timeout.run((signal) => fetch(signal), {
      error: new MemoizeTimeoutError(flightTimeout),
      timeout: flightTimeout,
    })

  if (!noCache) {
    const raw = await store.get(key)
    // Old cache entries used ambiguous untagged bigint strings. Refresh them
    // instead of guessing whether a matching string was originally a bigint.
    if (raw?.startsWith(jsonCodecPrefix)) {
      try {
        return jsonDecode(raw) as value
      } catch {
        // Corrupt entries are treated as misses so the next call refreshes.
      }
    }
  }

  const flights = flightsFor(store)
  const existing = flights.get(key)
  if (existing) return existing as Promise<value>

  // Coalesce concurrent misses for this process. The store remains the cache
  // boundary; this only prevents a local cold-key stampede while the first
  // fetch is still in flight.
  let mutating = noCache
  let settled = false
  const work = async (signal?: AbortSignal) => {
    if (noCache) {
      try {
        await store.delete(key)
      } finally {
        mutating = false
      }
    }
    signal?.throwIfAborted()
    mutating = false
    const value = await fetch(signal)
    signal?.throwIfAborted()
    if (value === undefined || options.shouldCache?.(value) === false) return value
    mutating = true
    if (store.type === 'cache') {
      // A lossy cache write cannot turn a successful origin read into an error.
      const persist = Promise.resolve()
        .then(() => store.put(key, jsonEncode(value), { ttl }))
        .catch(() => {})
      if (waitUntil)
        try {
          waitUntil(persist)
          mutating = false
          return value
        } catch {}
      try {
        await persist
      } finally {
        mutating = false
      }
      return value
    }
    try {
      await store.put(key, jsonEncode(value), { ttl })
    } finally {
      mutating = false
    }
    return value
  }
  let flight!: Promise<value>
  const release = () => {
    // A canceled fetch cannot mutate the cache after its next abort check.
    // Non-cancelable cache mutations retain ownership until they settle.
    if ((settled || !mutating) && flights.get(key) === flight) flights.delete(key)
  }
  flight = Timeout.run(
    async (signal) => {
      signal.addEventListener('abort', release, { once: true })
      try {
        return await work(signal)
      } finally {
        settled = true
        release()
      }
    },
    {
      error: new MemoizeTimeoutError(flightTimeout),
      timeout: flightTimeout,
    },
  )
  flights.set(key, flight)
  return flight
}

export declare namespace memoize {
  /** Options for {@link memoize}. */
  type Options<value = unknown> = {
    /** Maximum shared cache-miss flight duration in milliseconds. Defaults to 15 seconds. */
    flightTimeout?: number | undefined
    /** Cache key. Callers are responsible for namespacing/versioning. */
    key: string
    /** Returns whether a fetched value should be stored. Single-flight callers still share it. */
    shouldCache?: ((value: value) => boolean) | undefined
    /** Store used to persist cache entries. */
    store: Store
    /** Time-to-live in milliseconds. */
    ttl: number
  }
}

const requestStores = new WeakMap<
  Store,
  {
    request: withRequest.Request
    store: Store
    waitUntil?: ((promise: Promise<unknown>) => void) | undefined
  }
>()
const flights = new WeakMap<Store, Map<string, Promise<unknown>>>()

function flightsFor(store: Store) {
  let inFlight = flights.get(store)
  if (!inFlight) {
    inFlight = new Map()
    flights.set(store, inFlight)
  }
  return inFlight
}

function durableObjectNamespace(
  target: durableObject.Namespace | durableObject.Storage,
): target is durableObject.Namespace {
  return typeof target === 'object' && target !== null && 'getByName' in target
}

function durableObjectStub(
  namespace: durableObject.Namespace,
  name: string | ((key: string) => string),
): durableObject.ReturnValue {
  // Durable Object stubs are request-scoped I/O objects in Workers, so
  // resolve the stub inside each store operation instead of caching it.
  const stub = (key: string) => namespace.getByName(typeof name === 'function' ? name(key) : name)

  return {
    async delete(key: string) {
      await durableObjectRetry(() => stub(key).delete(key))
    },
    async deleteIf(key: string, expected: null | string) {
      const encoded = expected === null ? null : durableObjectEncode(expected)
      return stub(key).deleteIf(key, encoded)
    },
    async get(key: string) {
      return durableObjectDecode(await durableObjectRetry(() => stub(key).get(key)))
    },
    // Native RPC increment: the read-modify-write runs inside the object
    // (one round trip, atomic) instead of the get + put fallback (two).
    async increment(key: string, options?: Store.PutOptions) {
      return durableObjectRetry(() => stub(key).increment(key, options))
    },
    async list(options?: Parameters<Store['list']>[0]) {
      // A sharded store scatters keys across objects, so no single object can
      // enumerate them.
      if (typeof name === 'function')
        throw new TypeError('cannot list a sharded Durable Object store')
      return durableObjectRetry(() => stub('').list(options))
    },
    async put(key: string, value: unknown, options?: Store.PutOptions) {
      await durableObjectRetry(() => stub(key).put(key, durableObjectEncode(value), options))
    },
    async swap(key: string, expected: null | string, next: string, options?: Store.PutOptions) {
      const encoded = expected === null ? null : durableObjectEncode(expected)
      return stub(key).swap(key, encoded, durableObjectEncode(next), options)
    },
    type: 'state' as const,
    async update<result>(key: string, fn: (current: unknown) => Change<unknown, result>) {
      for (let attempt = 0; attempt < updateAttempts; attempt++) {
        const current = await durableObjectRetry(() => stub(key).get(key))
        const change = fn(durableObjectDecode(current))
        if (change.op === 'noop') return change.result

        // Retrying an ambiguous CAS can apply the transform twice.
        const changed =
          change.op === 'set'
            ? await stub(key).swap(key, current, durableObjectEncode(change.value))
            : await stub(key).deleteIf(key, current)
        if (changed) return change.result
      }
      throw new UpdateContentionError(key, updateAttempts)
    },
  } as durableObject.ReturnValue
}

/**
 * Runs a Durable Object stub call, retrying once when workerd marks the error
 * retryable (transient transport failures, e.g. "Network connection lost.")
 * and not overloaded. Callers pass a thunk so the retry resolves a fresh
 * stub: a stub that threw is disconnected.
 */
async function durableObjectRetry<value>(call: () => Promise<value>): Promise<value> {
  try {
    return await call()
  } catch (error) {
    const { overloaded, retryable } = (error ?? {}) as {
      overloaded?: boolean | undefined
      retryable?: boolean | undefined
    }
    if (retryable !== true || overloaded === true) throw error
    return call()
  }
}

const jsonCodecRoot = 'tempo-api:store:json:'
const jsonCodecVersion = 'v2'
const jsonCodecPrefix = `${jsonCodecRoot}${jsonCodecVersion}:`
const jsonValuePrefix = 'tempo-api:store:value:'
const legacyDurableObjectCodecPrefix = `${jsonCodecRoot}v1:`

function jsonEncode(value: unknown): string {
  const encoded = JSON.stringify(value, (_key, item) => {
    if (typeof item === 'bigint') return `${jsonValuePrefix}bigint:${item}`
    if (typeof item === 'string' && item.startsWith(jsonValuePrefix))
      return `${jsonValuePrefix}string:${item}`
    return item
  })
  if (encoded === undefined) throw new TypeError('Store value is not serializable')
  return `${jsonCodecPrefix}${encoded}`
}

function jsonDecode(value: string): unknown {
  return JSON.parse(value.slice(jsonCodecPrefix.length), (_key, item) => {
    if (typeof item !== 'string' || !item.startsWith(jsonValuePrefix)) return item
    if (item.startsWith(`${jsonValuePrefix}string:`))
      return item.slice(`${jsonValuePrefix}string:`.length)
    if (item.startsWith(`${jsonValuePrefix}bigint:`)) {
      const bigint = item.slice(`${jsonValuePrefix}bigint:`.length)
      if (!/^-?\d+$/.test(bigint)) throw new TypeError('Store bigint is malformed')
      return BigInt(bigint)
    }
    throw new TypeError('Store value tag is unsupported')
  })
}

function durableObjectEncode(value: unknown): string {
  if (typeof value === 'string' && !value.startsWith(jsonCodecRoot)) return value
  try {
    return jsonEncode(value)
  } catch (cause) {
    throw new TypeError('Durable Object store value is not serializable', {
      cause,
    })
  }
}

function durableObjectDecode(value: null | string): unknown {
  if (value === null || !value.startsWith(jsonCodecRoot)) return value
  if (value.startsWith(jsonCodecPrefix)) return jsonDecode(value)
  if (value.startsWith(legacyDurableObjectCodecPrefix))
    return Json.parse(value.slice(legacyDurableObjectCodecPrefix.length))
  const [version] = value.slice(jsonCodecRoot.length).split(':')
  throw new DurableObjectCodecError(version)
}

function cloudflareKvExpirationTtl(ttl: number | undefined) {
  if (ttl === undefined) return undefined
  if (ttl <= 0) return 0

  // Workers KV expects TTLs in seconds and rejects values under 60 seconds.
  const seconds = Math.ceil(ttl / 1_000)
  if (seconds < 60) throw new RangeError('Cloudflare KV expirationTtl must be at least 60 seconds')
  return seconds
}

const durableObjectValueType = 'tempo-api:store:v1'

type DurableObjectValue = {
  expiresAt?: number | undefined
  type: typeof durableObjectValueType
  value: string
}

function durableObjectValue(value: string, options: Store.PutOptions): DurableObjectValue | string {
  if (options.ttl === undefined) return value

  // Durable Object storage does not expose per-key TTL in the current Worker
  // API, so this adapter stores expirations in-band and evicts lazily on read.
  return { expiresAt: Date.now() + options.ttl, type: durableObjectValueType, value }
}

async function durableObjectRead(
  storage: durableObject.Storage,
  key: string,
  value: unknown,
): Promise<DurableObjectValue | { value: string } | null> {
  if (typeof value === 'string') return { value }
  if (!isDurableObjectValue(value)) return null
  if (value.expiresAt !== undefined && value.expiresAt <= Date.now()) {
    await storage.delete(key)
    return null
  }
  return value
}

function isDurableObjectValue(value: unknown): value is DurableObjectValue {
  if (!value || typeof value !== 'object') return false
  return (
    'type' in value &&
    value.type === durableObjectValueType &&
    'value' in value &&
    typeof value.value === 'string' &&
    (!('expiresAt' in value) ||
      value.expiresAt === undefined ||
      typeof value.expiresAt === 'number')
  )
}

/** Thrown when a shared memoized cache-miss flight exceeds its configured deadline. */
export class MemoizeTimeoutError extends Error {
  override name = 'Store.MemoizeTimeoutError'

  constructor(timeout: number) {
    super(`Memoized cache-miss flight timed out after ${timeout}ms.`)
  }
}

/** Thrown when {@link update} exhausts its compare-and-swap attempts. */
export class UpdateContentionError extends Error {
  override name = 'Store.UpdateContentionError'

  constructor(key: string, attempts: number) {
    super(`Store update on "${key}" lost ${attempts} compare-and-swap attempts.`)
  }
}

/** Thrown when Durable Object state uses an unsupported codec. */
export class DurableObjectCodecError extends Error {
  override name = 'Store.DurableObjectCodecError'

  constructor(version: unknown) {
    super(`Unsupported Durable Object store codec version: ${String(version)}`)
  }
}