zoriya/v10
1
0
Fork 0
mirror of https://github.com/zoriya/v10.git synced 2026-08-16 02:45:09 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(store): queue task refactor (#287)

Browse Source
This commit is contained in:
rahim
2026-01-05 16:24:18 +11:00
committed by GitHub
parent 44cd824f3b
commit 6a7acdabfe
9 changed files with 941 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.claude/plans/store-bindings.md
+81 -157
View File
@@ -1,5 +1,7 @@
# Store React/DOM Bindings # Store React/DOM Bindings
> **For AI agents:** When marking a phase as complete, remove detailed API specs and replace with a PR reference (e.g., "Refer to PR #XXX for implementation details"). The PR is the source of truth for completed work.
## Goal ## Goal
Implement React and DOM bindings for Video.js 10's store, enabling: Implement React and DOM bindings for Video.js 10's store, enabling:
@@ -11,133 +13,53 @@ Implement React and DOM bindings for Video.js 10's store, enabling:
## Key Decisions ## Key Decisions
| Decision | Resolution | | Decision | Resolution |
| ------------------- | ------------------------------------------------------------------------------------- | | ------------------- | ----------------------------------------------------------------------------------- |
| Store creation | `createStore({ slices, displayName? })` - types inferred from slices | | Store creation | `createStore({ slices, displayName? })` - types inferred from slices |
| Hook naming | `useStore`, `useSelector`, `useRequest`, `usePending`, `useMutation`, `useOptimistic` | | Hook naming | `useStore`, `useSelector`, `useRequest`, `useTasks`, `useMutation`, `useOptimistic` |
| Controller naming | `SelectorController`, `RequestController`, `PendingController`, etc | | Controller naming | `SelectorController`, `RequestController`, `TasksController`, etc |
| Selector hook | `useSelector(selector)` - requires selector (Redux-style) | | Selector hook | `useSelector(selector)` - requires selector (Redux-style) |
| Store hook | `useStore()` - returns store instance | | Store hook | `useStore()` - returns store instance |
| Request hook | `useRequest()` or `useRequest(r => r.foo)` - full map or single request | | Request hook | `useRequest()` or `useRequest(r => r.foo)` - full map or single request |
| Pending hook | `usePending()` - returns `store.queue.pending` (reactive) | <<<<<<< Updated upstream
| Mutation hook | `useMutation(r => r.foo)` - status tracking (isPending, isError, error) | | Tasks hook | `useTasks()` - returns `store.queue.tasks` (reactive) |
| Optimistic hook | `useOptimistic(r => r.foo, s => s.bar)` - optimistic value + status | =======
| Settled state | Core Queue tracks last result/error per key, cleared on next request | | Tasks hook | `useTasks()` - returns `store.queue.tasks` (reactive, full lifecycle) |
| Base hooks | All take store as first arg: `useSelector(store, sel)`, etc | >>>>>>> Stashed changes
| createStore hooks | Returns all hooks including `useMutation` and `useOptimistic` | | Mutation hook | `useMutation(r => r.foo)` - status tracking (isPending, isError, error) |
| Slice hook return | `{ state, request, isAvailable }` - state/request null when unavailable | | Optimistic hook | `useOptimistic(r => r.foo, s => s.bar)` - optimistic value + status |
| Skin exports | `Provider`, `Skin`, `extendConfig` | | Settled state | Core Queue tracks last result/error per key, cleared on next request |
| Slice namespace | `export * as media` → `media.playback` | | Base hooks | All take store as first arg: `useSelector(store, sel)`, etc |
| Video component | Generic, exported from `@videojs/react` (not from skins) | | createStore hooks | Returns all hooks including `useMutation` and `useOptimistic` |
| Lit mixins | `StoreMixin` (combined), `StoreProviderMixin`, `StoreAttachMixin` | | Slice hook return | `{ state, request, isAvailable }` - state/request null when unavailable |
| Primitives context | `useStoreContext()` internal hook for primitive UI components | | Skin exports | `Provider`, `Skin`, `extendConfig` |
| displayName | For React DevTools component naming | | Slice namespace | `export * as media` → `media.playback` |
| Component types | Namespace pattern: `Skin.Props` via `namespace Skin { export type Props }` | | Video component | Generic, exported from `@videojs/react` (not from skins) |
| Element define | `FrostedSkinElement.define(tagName, { mixins })` for declarative setup | | Lit mixins | `StoreMixin` (combined), `StoreProviderMixin`, `StoreAttachMixin` |
| Config extension | `extendConfig()` uses `uniqBy` + `composeCallbacks` from utils | | Primitives context | `useStoreContext()` internal hook for primitive UI components |
| Provider resolution | Isolated by default; `inherit` prop to use parent store from context | | displayName | For React DevTools component naming |
| Store instance | `create()` method for imperative store creation | | Component types | Namespace pattern: `Skin.Props` via `namespace Skin { export type Props }` |
| Package structure | `store/react` and `store/lit` (no `store/dom`) | | Element define | `FrostedSkinElement.define(tagName, { mixins })` for declarative setup |
| Config extension | `extendConfig()` uses `uniqBy` + `composeCallbacks` from utils |
| Provider resolution | Isolated by default; `inherit` prop to use parent store from context |
| Store instance | `create()` method for imperative store creation |
| Package structure | `store/react` and `store/lit` (no `store/dom`) |
--- ---
## Phase 0: Core Utilities [DONE] ## Phase 0: Core Utilities [DONE]
> Implemented in PR #283. > Refer to [PR #283](https://github.com/videojs/v10/pull/283) for implementation details.
- `uniqBy` - `packages/utils/src/array/uniq-by.ts` Added `uniqBy`, `composeCallbacks` utilities and `extendConfig` for store.
- `composeCallbacks` - `packages/utils/src/function/compose-callbacks.ts`
- `extendConfig` - `packages/store/src/core/extend-config.ts`
--- ---
## Phase 0.5: Queue Task Refactor ## Phase 0.5: Queue Task Refactor [DONE]
Refactor Queue to use a unified `tasks` map with status discriminator. This enables `useMutation` and `useOptimistic` hooks to track request lifecycle. > Refer to [PR #287](https://github.com/videojs/v10/pull/287) for implementation details.
**File:** `packages/store/src/core/queue.ts` Refactored Queue to use unified `tasks` map with status discriminator (`PendingTask | SuccessTask | ErrorTask`). Added `tryCatch` utility to `@videojs/utils/function`.
### Task Types (Discriminated Union)
```typescript
// Base fields shared by all task states
interface TaskBase<Key, Input> {
id: symbol;
name: string;
key: Key;
input: Input;
startedAt: number;
meta: RequestMeta | null;
}
// Pending - request in flight
interface PendingTask<Key, Input> extends TaskBase<Key, Input> {
status: 'pending';
abort: AbortController;
}
// Success - completed successfully
interface SuccessTask<Key, Input, Output> extends TaskBase<Key, Input> {
status: 'success';
settledAt: number;
duration: number;
output: Output;
}
// Error - failed or cancelled
interface ErrorTask<Key, Input> extends TaskBase<Key, Input> {
status: 'error';
settledAt: number;
duration: number;
error: unknown;
cancelled: boolean; // true if aborted, false if actual error
}
// Union types
type Task<Key, Input, Output> = PendingTask<Key, Input> | SuccessTask<Key, Input, Output> | ErrorTask<Key, Input>;
type SettledTask<Key, Input, Output> = SuccessTask<Key, Input, Output> | ErrorTask<Key, Input>;
```
### Queue API
```typescript
export class Queue<Tasks extends TaskRecord> {
// Single source of truth - one task per key (pending OR settled)
get tasks(): Readonly<TasksRecord<Tasks>>;
// Clear settled task for a key (no-op if pending)
reset(key: keyof Tasks): void;
// Subscribe to task changes
subscribe(listener: (tasks: TasksRecord<Tasks>) => void): () => void;
}
```
### Lifecycle
1. `enqueue()` → task added with `status: 'pending'`
2. Task completes → same entry updated to `status: 'success'` or `status: 'error'`
3. New request for same key → replaces previous (pending aborted, settled cleared)
4. `reset(key)` → removes settled task
### Usage
```typescript
const task = queue.tasks.changeVolume;
// TypeScript narrows based on status
if (task?.status === 'pending') {
task.abort; // available
}
if (task?.status === 'success') {
task.output; // available
}
if (task?.status === 'error') {
task.error; // available
task.cancelled; // true if aborted
}
```
--- ---
@@ -191,8 +113,8 @@ export function StoreContextProvider({ store, children }: { store: AnyStore; chi
**File:** `packages/store/src/react/create-store.ts` **File:** `packages/store/src/react/create-store.ts`
```typescript ```typescript
import type { ReactNode } from 'react';
import type { AnySlice, InferSliceTarget, StoreConfig } from '../core'; import type { AnySlice, InferSliceTarget, StoreConfig } from '../core';
import type { ReactNode } from 'react';
import { useEffect, useState } from 'react'; import { useEffect, useState } from 'react';
@@ -221,7 +143,7 @@ export interface CreateStoreResult<Slices extends AnySlice[]> {
(): UnionSliceRequests<Slices>; (): UnionSliceRequests<Slices>;
<T>(selector: (requests: UnionSliceRequests<Slices>) => T): T; <T>(selector: (requests: UnionSliceRequests<Slices>) => T): T;
}; };
usePending: () => PendingRecord<UnionSliceTasks<Slices>>; useTasks: () => TasksRecord<UnionSliceTasks<Slices>>;
useMutation: <K extends keyof UnionSliceRequests<Slices>>( useMutation: <K extends keyof UnionSliceRequests<Slices>>(
selector: (requests: UnionSliceRequests<Slices>) => UnionSliceRequests<Slices>[K] selector: (requests: UnionSliceRequests<Slices>) => UnionSliceRequests<Slices>[K]
) => MutationResult<UnionSliceRequests<Slices>[K]>; ) => MutationResult<UnionSliceRequests<Slices>[K]>;
@@ -241,9 +163,9 @@ export function createStore<Slices extends AnySlice[]>(config: CreateStoreConfig
**File:** `packages/store/src/react/types.ts` **File:** `packages/store/src/react/types.ts`
```typescript ```typescript
export type SliceResult<S extends AnySlice> export type SliceResult<S extends AnySlice> =
= | { state: InferSliceState<S>; request: InferSliceRequests<S>; isAvailable: true } | { state: InferSliceState<S>; request: InferSliceRequests<S>; isAvailable: true }
| { state: null; request: null; isAvailable: false }; | { state: null; request: null; isAvailable: false };
export interface MutationResult<Request extends (...args: any[]) => any> { export interface MutationResult<Request extends (...args: any[]) => any> {
/** Trigger the request */ /** Trigger the request */
@@ -285,7 +207,7 @@ export function useSelector<S extends AnyStore, T>(store: S, selector: (state: I
export function useRequest<S extends AnyStore>(store: S): InferStoreRequests<S>; export function useRequest<S extends AnyStore>(store: S): InferStoreRequests<S>;
export function useRequest<S extends AnyStore, T>(store: S, selector: (requests: InferStoreRequests<S>) => T): T; export function useRequest<S extends AnyStore, T>(store: S, selector: (requests: InferStoreRequests<S>) => T): T;
export function usePending<S extends AnyStore>(store: S): PendingRecord<InferStoreTasks<S>>; export function useTasks<S extends AnyStore>(store: S): TasksRecord<InferStoreTasks<S>>;
export function useMutation<S extends AnyStore, R extends (...args: any[]) => any>( export function useMutation<S extends AnyStore, R extends (...args: any[]) => any>(
store: S, store: S,
@@ -332,7 +254,7 @@ export function useOptimistic<S extends AnyStore, R extends (...args: any[]) =>
- `useStore()`: Returns typed store from `useStoreContext()` - `useStore()`: Returns typed store from `useStoreContext()`
- `useSelector(selector)`: Uses `useSyncExternalStore` with selector - `useSelector(selector)`: Uses `useSyncExternalStore` with selector
- `useRequest()`: Returns stable `store.request` from context - `useRequest()`: Returns stable `store.request` from context
- `usePending()`: Subscribes to `store.queue`, returns `queue.pending` - `useTasks()`: Subscribes to `store.queue`, returns `queue.tasks`
- `useSlice(slice)`: Returns `{ state, request, isAvailable }` with null narrowing - `useSlice(slice)`: Returns `{ state, request, isAvailable }` with null narrowing
### 1.6 Exports ### 1.6 Exports
@@ -344,7 +266,7 @@ export function useOptimistic<S extends AnyStore, R extends (...args: any[]) =>
export { useStoreContext } from './context'; export { useStoreContext } from './context';
export { createStore } from './create-store'; export { createStore } from './create-store';
// Base hooks for testing/advanced use (all take store as first arg) // Base hooks for testing/advanced use (all take store as first arg)
export { useMutation, useOptimistic, usePending, useRequest, useSelector } from './hooks'; export { useMutation, useOptimistic, useTasks, useRequest, useSelector } from './hooks';
export type { CreateStoreConfig, CreateStoreResult, MutationResult, OptimisticResult, SliceResult } from './types'; export type { CreateStoreConfig, CreateStoreResult, MutationResult, OptimisticResult, SliceResult } from './types';
``` ```
@@ -389,9 +311,9 @@ class MySkin extends HTMLElement {
**File:** `packages/store/src/lit/create-store.ts` **File:** `packages/store/src/lit/create-store.ts`
```typescript ```typescript
import type { AnySlice, InferSliceTarget, StoreConfig } from '../core';
import type { Context } from '@lit/context'; import type { Context } from '@lit/context';
import type { ReactiveControllerHost } from '@lit/reactive-element'; import type { ReactiveControllerHost } from '@lit/reactive-element';
import type { AnySlice, InferSliceTarget, StoreConfig } from '../core';
import { createContext } from '@lit/context'; import { createContext } from '@lit/context';
@@ -458,10 +380,10 @@ export class RequestController<S extends AnyStore, T> implements ReactiveControl
get value(): T; get value(): T;
} }
// PendingController - like usePending(store) // TasksController - like useTasks(store)
export class PendingController<S extends AnyStore> implements ReactiveController { export class TasksController<S extends AnyStore> implements ReactiveController {
constructor(host: ReactiveControllerHost, store: S); constructor(host: ReactiveControllerHost, store: S);
get value(): PendingRecord<InferStoreTasks<S>>; get value(): TasksRecord<InferStoreTasks<S>>;
} }
// MutationController - like useMutation(store, selector) // MutationController - like useMutation(store, selector)
@@ -500,7 +422,7 @@ export class OptimisticController<
export { export {
MutationController, MutationController,
OptimisticController, OptimisticController,
PendingController, TasksController,
RequestController, RequestController,
SelectorController, SelectorController,
} from './controllers'; } from './controllers';
@@ -928,16 +850,16 @@ function App() {
} }
function MyCustomControls() { function MyCustomControls() {
const currentTime = useSelector(s => s.currentTime); const currentTime = useSelector((s) => s.currentTime);
const seek = useRequest(r => r.seek); const seek = useRequest((r) => r.seek);
return <button onClick={() => seek(0)}>Restart ({currentTime}s)</button>; return <button onClick={() => seek(0)}>Restart ({currentTime}s)</button>;
} }
// With mutation status tracking // With mutation status tracking
function PlayButton() { function PlayButton() {
const paused = useSelector(s => s.paused); const paused = useSelector((s) => s.paused);
const { mutate: play, isPending } = useMutation(r => r.play); const { mutate: play, isPending } = useMutation((r) => r.play);
const { mutate: pause } = useMutation(r => r.pause); const { mutate: pause } = useMutation((r) => r.pause);
return ( return (
<button onClick={() => (paused ? play() : pause())} disabled={isPending}> <button onClick={() => (paused ? play() : pause())} disabled={isPending}>
@@ -949,8 +871,8 @@ function PlayButton() {
// With optimistic updates // With optimistic updates
function VolumeSlider() { function VolumeSlider() {
const { value, setValue, isPending, isError } = useOptimistic( const { value, setValue, isPending, isError } = useOptimistic(
r => r.changeVolume, (r) => r.changeVolume,
s => s.volume (s) => s.volume
); );
return ( return (
@@ -958,7 +880,7 @@ function VolumeSlider() {
<input <input
type="range" type="range"
value={value} value={value}
onChange={e => setValue(Number(e.target.value))} onChange={(e) => setValue(Number(e.target.value))}
style={{ opacity: isPending ? 0.5 : 1 }} style={{ opacity: isPending ? 0.5 : 1 }}
/> />
{isError && <span>Failed to change volume</span>} {isError && <span>Failed to change volume</span>}
@@ -970,10 +892,10 @@ function VolumeSlider() {
### React: Pre-created store instance (imperative access) ### React: Pre-created store instance (imperative access)
```tsx ```tsx
import { createStore, media, Video } from '@videojs/react';
import { useState } from 'react'; import { useState } from 'react';
import { createStore, media, Video } from '@videojs/react';
const { Provider, create, useSelector } = createStore({ const { Provider, create, useSelector } = createStore({
slices: [media.playback], slices: [media.playback],
}); });
@@ -1186,21 +1108,22 @@ packages/html/src/
- `uniqBy`, `composeCallbacks` utilities ✓ - `uniqBy`, `composeCallbacks` utilities ✓
- `extendConfig` ✓ - `extendConfig` ✓
2. **Phase 0.5**: Queue Task Refactor 2. **Phase 0.5**: Queue Task Refactor **[DONE - PR #287]**
- Unified `tasks` map with status discriminator - Unified `tasks` map with status discriminator ✓
- `PendingTask`, `SuccessTask`, `ErrorTask` types - `PendingTask`, `SuccessTask`, `ErrorTask` types ✓
- `reset(key)` method - `reset(key)` method ✓
- Update existing tests - Update existing tests ✓
- Added `tryCatch` utility to `@videojs/utils/function` ✓
3. **Phase 1**: React Bindings (basic) 3. **Phase 1**: React Bindings (basic)
- Shared context, `useStoreContext` - Shared context, `useStoreContext`
- `createStore()` with `inherit` prop - `createStore()` with `inherit` prop
- `useStore`, `useSelector`, `useRequest`, `usePending` - `useStore`, `useSelector`, `useRequest`, `useTasks`
- `Video` component, package exports - `Video` component, package exports
4. **Phase 2**: Lit Bindings (basic) 4. **Phase 2**: Lit Bindings (basic)
- `createStore()` with mixins - `createStore()` with mixins
- `SelectorController`, `RequestController`, `PendingController` - `SelectorController`, `RequestController`, `TasksController`
- `@lit/context` integration - `@lit/context` integration
5. **Phase 3**: Mutation Hooks/Controllers 5. **Phase 3**: Mutation Hooks/Controllers
@@ -1235,7 +1158,7 @@ import { readdirSync } from 'node:fs';
// Dynamically gather define/ entries // Dynamically gather define/ entries
const defineEntries = readdirSync('src/define') const defineEntries = readdirSync('src/define')
.filter(f => f.endsWith('.ts')) .filter((f) => f.endsWith('.ts'))
.reduce( .reduce(
(acc, f) => { (acc, f) => {
const name = f.replace('.ts', ''); const name = f.replace('.ts', '');
@@ -1363,23 +1286,24 @@ PR #283: Core Utilities [DONE]
├── extendConfig (store/core) ✓ ├── extendConfig (store/core) ✓
└── Tests ✓ └── Tests ✓
PR A: Queue Task Refactor PR #287: Queue Task Refactor [DONE]
├── Unified Task type with status discriminator ├── Unified Task type with status discriminator ✓
├── PendingTask, SuccessTask, ErrorTask ├── PendingTask, SuccessTask, ErrorTask ✓
├── Single `tasks` map, `reset(key)` method ├── Single `tasks` map, `reset(key)` method ✓
├── Update tests ├── Update tests ✓
├── Added tryCatch utility ✓
└── Closes #285 └── Closes #285
PR B: React Bindings (basic) PR B: React Bindings (basic)
├── createStore, Provider, useStore ├── createStore, Provider, useStore
├── useSelector, useRequest, usePending ├── useSelector, useRequest, useTasks
├── Video component ├── Video component
├── References #218 ├── References #218
└── Closes #229 └── Closes #229
PR C: Lit Bindings (basic) PR C: Lit Bindings (basic)
├── createStore with mixins ├── createStore with mixins
├── SelectorController, RequestController, PendingController ├── SelectorController, RequestController, TasksController
├── References #218 ├── References #218
└── Closes #230 └── Closes #230
@@ -1409,9 +1333,9 @@ PR G: Skins
### Dependency Graph ### Dependency Graph
``` ```
PR #283 ───> PR A ───> PR B ───> PR D ───> PR E ───> PR G PR #283 ───> PR #287 ───> PR B ───> PR D ───> PR E ───> PR G
└──> PR C ──────────────────────────┘ └──> PR C ──────────────────────────┘
└──> PR F ──────────────────────────┘ └──> PR F ──────────────────────────┘
``` ```
PRs are sequential. PR B, C, F can technically parallel after PR A, but we'll do them sequentially for easier review. PRs are sequential. PR B, C, F can technically parallel after PR A, but we'll do them sequentially for easier review.
CLAUDE.md
+83 -5
View File
@@ -94,11 +94,14 @@ pnpm clean
## Dev Workflow ## Dev Workflow
1. Make changes. 1. Make changes.
2. Typecheck, fix all issues. 2. If you added/changed **exported types** in a package, run `pnpm -F <pkg> build` first.
3. Run test/s, fix all issues. If there are no tests add them. - `pnpm typecheck` uses TypeScript project references against **built** `.d.ts` files.
4. Lint file/s, fix all issues. - New/changed types won't be visible until `tsdown` builds them.
5. Run build/s, fix all errors. 3. Typecheck, fix all issues.
6. Before creating a PR `pnpm test`. 4. Run test/s, fix all issues. If there are no tests add them.
5. Lint file/s, fix all issues.
6. Run build/s, fix all errors.
7. Before creating a PR `pnpm test`.
Be efficient when running operations, see "Common Root Commands". Be efficient when running operations, see "Common Root Commands".
@@ -158,3 +161,78 @@ When generating or editing code in this repository, follow these rules to ensure
- Use semantic commit messages (enforced by `commitlint`). - Use semantic commit messages (enforced by `commitlint`).
- One focused change per commit—no mixed updates. - One focused change per commit—no mixed updates.
- Breaking changes use `!`. - Breaking changes use `!`.
## Code Rules
Prefer existing utilities over inline implementations:
| Instead of | Use |
| ------------------------- | ------------------------------------------------------ |
| `x === undefined` | `isUndefined(x)` from `@videojs/utils/predicate` |
| `x === null` | `isNull(x)` from `@videojs/utils/predicate` |
| `typeof x === 'function'` | `isFunction(x)` from `@videojs/utils/predicate` |
| `typeof x === 'string'` | `isString(x)` from `@videojs/utils/predicate` |
Before writing new helpers, check `@videojs/utils` for existing utilities.
### Naming Conventions
| Pattern | Prefix | Example |
| ------------------- | ---------- | -------------------------------- |
| Type inference | `Infer*` | `InferSliceState<S>` |
| Type resolution | `Resolve*` | `ResolveRequestHandler<R>` |
| Type constraint | `Ensure*` | `EnsureTaskRecord<T>` |
| Union type helpers | `Union*` | `UnionSliceState<Slices>` |
| Default loose types | `Default*` | `DefaultTaskRecord` |
| Type guards | `is*` | `isStoreError(error)` |
| Factory functions | `create*` | `createQueue()`, `createSlice()` |
### Type Guards
Always return `value is Type` for proper type narrowing:
```ts
function isStoreError(error: unknown): error is StoreError {
return error instanceof StoreError;
}
```
### Subscribe Pattern
Subscriptions return an unsubscribe function:
```ts
subscribe(listener: Listener): () => void {
this.#subscribers.add(listener);
return () => this.#subscribers.delete(listener);
}
```
### Optional Key Parameter
Methods that operate on one or all items use optional key:
```ts
// If key provided: operate on that item
// If no key: operate on all items
reset(key?: keyof Tasks): void {
if (!isUndefined(key)) {
// Single item
return;
}
// All items
}
```
### Destroy Pattern
Guard re-entry, set flag first, cleanup in order:
```ts
destroy(): void {
if (this.#destroyed) return;
this.#destroyed = true;
this.abort();
this.#subscribers.clear();
}
```
packages/store/README.md
+29 -21
View File
@@ -263,26 +263,26 @@ const unsubscribe = store.subscribe((state) => {
// Single value - only fires when volume changes // Single value - only fires when volume changes
store.subscribe( store.subscribe(
s => s.volume, (s) => s.volume,
volume => console.log('Volume:', volume) (volume) => console.log('Volume:', volume)
); );
// Multiple values - auto-optimized with key-based subscription // Multiple values - auto-optimized with key-based subscription
store.subscribe( store.subscribe(
s => ({ volume: s.volume, muted: s.muted }), (s) => ({ volume: s.volume, muted: s.muted }),
({ volume, muted }) => updateAudioUI(volume, muted) ({ volume, muted }) => updateAudioUI(volume, muted)
); );
// Derived value // Derived value
store.subscribe( store.subscribe(
s => Math.round(s.volume * 100), (s) => Math.round(s.volume * 100),
percent => console.log(`${percent}%`) (percent) => console.log(`${percent}%`)
); );
// Custom equality function // Custom equality function
store.subscribe( store.subscribe(
s => s.playlist, (s) => s.playlist,
playlist => renderPlaylist(playlist), (playlist) => renderPlaylist(playlist),
{ equalityFn: shallowEqual } { equalityFn: shallowEqual }
); );
``` ```
@@ -542,17 +542,17 @@ const store = createStore({
], ],
queue: createQueue({ queue: createQueue({
// Default scheduler for requests without schedule // Default scheduler for requests without schedule
scheduler: flush => queueMicrotask(flush), scheduler: (flush) => queueMicrotask(flush),
// Lifecycle hooks // Lifecycle hooks
onDispatch: (request) => { onDispatch: (request) => {
console.log('Started:', request.name); console.log('Started:', request.name);
}, },
onSettled: (request, { status, duration }) => { onSettled: (task) => {
analytics.track(request.name, { analytics.track(task.name, {
status, status: task.status,
duration, duration: task.settledAt - task.startedAt,
}); });
}, },
}), }),
@@ -564,21 +564,29 @@ const store = createStore({
```ts ```ts
const queue = store.queue; // accessed on the store const queue = store.queue; // accessed on the store
queue.queued; // object of requests waiting to execute queue.queued; // tasks waiting to execute
queue.pending; // object of requests currently executing queue.tasks; // task lifecycle map (pending/success/error)
// Check if a task is pending or queued // Check task status
queue.isPending('playback'); // true if currently executing queue.isPending('playback'); // true if currently executing
queue.isQueued('seek'); // true if waiting to execute queue.isQueued('seek'); // true if waiting to execute
queue.isSettled('seek'); // true if completed (success or error)
queue.dequeue('seek'); // remove from queue without executing // Cancel queued tasks (waiting to execute)
queue.clear(); // clear all queued queue.cancel('seek'); // cancel specific
queue.cancel(); // cancel all queued
// Abort tasks (queued + executing)
queue.abort('playback'); // abort specific
queue.abort(); // abort all
// Clear settled tasks (success/error results)
queue.reset('seek'); // clear specific
queue.reset(); // clear all settled
// Execute queued tasks immediately
queue.flush(); // execute all queued now queue.flush(); // execute all queued now
queue.flush('playback'); // execute specific key now queue.flush('playback'); // execute specific key now
queue.abort('playback'); // abort executing request
queue.abortAll(); // abort all executing
``` ```
### Direct Queue Usage ### Direct Queue Usage
@@ -620,7 +628,7 @@ const store = createStore({
slices: [ slices: [
/* ... */ /* ... */
], ],
state: initial => new VueStateAdapter(initial), state: (initial) => new VueStateAdapter(initial),
}); });
``` ```
packages/store/src/core/queue.ts
+201 -79
View File
@@ -1,5 +1,6 @@
import type { Request, RequestMeta } from './request'; import type { Request, RequestMeta } from './request';
import { tryCatch } from '@videojs/utils/function';
import { isFunction, isUndefined } from '@videojs/utils/predicate'; import { isFunction, isUndefined } from '@videojs/utils/predicate';
import { StoreError } from './errors'; import { StoreError } from './errors';
@@ -37,18 +38,62 @@ export type DefaultTaskRecord = Record<TaskKey, Request<unknown, unknown>>;
export type EnsureTaskRecord<T> = T extends TaskRecord ? T : never; export type EnsureTaskRecord<T> = T extends TaskRecord ? T : never;
/** /**
* Pending task info. * Base fields shared by all task states.
*/ */
export interface PendingTask<Key extends TaskKey = TaskKey, Input = unknown> { export interface TaskBase<Key extends TaskKey = TaskKey, Input = unknown> {
id: symbol; id: symbol;
name: string; name: string;
key: Key; key: Key;
input: Input; input: Input;
startedAt: number; startedAt: number;
abort: AbortController;
meta: RequestMeta | null; meta: RequestMeta | null;
} }
/**
* Pending task - request in flight.
*/
export interface PendingTask<Key extends TaskKey = TaskKey, Input = unknown> extends TaskBase<Key, Input> {
status: 'pending';
abort: AbortController;
}
/**
* Success task - completed successfully.
*/
export interface SuccessTask<Key extends TaskKey = TaskKey, Input = unknown, Output = unknown> extends TaskBase<
Key,
Input
> {
status: 'success';
settledAt: number;
output: Output;
}
/**
* Error task - failed or cancelled.
*/
export interface ErrorTask<Key extends TaskKey = TaskKey, Input = unknown> extends TaskBase<Key, Input> {
status: 'error';
settledAt: number;
error: unknown;
cancelled: boolean;
}
/**
* Task with status discriminator.
*/
export type Task<Key extends TaskKey = TaskKey, Input = unknown, Output = unknown>
= | PendingTask<Key, Input>
| SuccessTask<Key, Input, Output>
| ErrorTask<Key, Input>;
/**
* Settled task (success or error).
*/
export type SettledTask<Key extends TaskKey = TaskKey, Input = unknown, Output = unknown>
= | SuccessTask<Key, Input, Output>
| ErrorTask<Key, Input>;
/** /**
* Context passed to task handler. * Context passed to task handler.
*/ */
@@ -90,13 +135,7 @@ export interface QueueConfig<Tasks extends TaskRecord = DefaultTaskRecord> {
/** Default scheduler when task has no schedule */ /** Default scheduler when task has no schedule */
scheduler?: TaskScheduler; scheduler?: TaskScheduler;
onDispatch?: <K extends keyof Tasks>(task: PendingTask<TaskKey<K>, Tasks[K]['input']>) => void; onDispatch?: <K extends keyof Tasks>(task: PendingTask<TaskKey<K>, Tasks[K]['input']>) => void;
onSettled?: <K extends keyof Tasks>( onSettled?: <K extends keyof Tasks>(task: SettledTask<TaskKey<K>, Tasks[K]['input'], Tasks[K]['output']>) => void;
task: PendingTask<TaskKey<K>, Tasks[K]['input']>,
result:
| { status: 'success'; duration: number; output: Tasks[K]['output'] }
| { status: 'cancelled'; error: unknown; duration: number }
| { status: 'error'; error: unknown; duration: number },
) => void;
} }
export interface QueuedTaskId<Key extends TaskKey = TaskKey> { export interface QueuedTaskId<Key extends TaskKey = TaskKey> {
@@ -112,16 +151,19 @@ export type QueuedRecord<Tasks extends TaskRecord> = {
[K in keyof Tasks]?: QueuedTask<TaskKey<K>>; [K in keyof Tasks]?: QueuedTask<TaskKey<K>>;
}; };
export type PendingRecord<Tasks extends TaskRecord> = { /**
[K in keyof Tasks]?: PendingTask<TaskKey<K>, Tasks[K]['input']>; * Map of task key -> task (pending, success, or error).
*/
export type TasksRecord<Tasks extends TaskRecord> = {
[K in keyof Tasks]?: Task<TaskKey<K>, Tasks[K]['input'], Tasks[K]['output']>;
}; };
/** /**
* Listener callback for pending state changes. * Listener callback for task state changes.
* *
* Called when tasks are dispatched or settled. * Called when tasks are dispatched, settled, or reset.
*/ */
export type QueueListener<Tasks extends TaskRecord> = (pending: PendingRecord<Tasks>) => void; export type QueueListener<Tasks extends TaskRecord> = (tasks: TasksRecord<Tasks>) => void;
// ---------------------------------------- // ----------------------------------------
// Schedulers // Schedulers
@@ -168,36 +210,27 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
readonly #subscribers = new Set<QueueListener<Tasks>>(); readonly #subscribers = new Set<QueueListener<Tasks>>();
#queued: QueuedRecord<Tasks> = {}; #queued: QueuedRecord<Tasks> = {};
#pending: PendingRecord<Tasks> = {}; #tasks: TasksRecord<Tasks> = {};
#destroyed = false; #destroyed = false;
constructor(config: QueueConfig<Tasks> = {}) { constructor(config: QueueConfig<Tasks> = {}) {
this.#scheduler = config.scheduler ?? microtask; this.#scheduler = config.scheduler ?? microtask;
// Wrap callbacks to catch errors and prevent breaking queue/scheduler // Wrap callbacks to catch errors and prevent breaking queue/scheduler
const safeCallback = <Args extends [PendingTask, ...unknown[]]>( const logError = (e: unknown) => console.error('[vjs-queue]', e);
callback: ((...args: Args) => unknown) | undefined, this.#onDispatch = tryCatch(config.onDispatch, logError);
) => { this.#onSettled = tryCatch(config.onSettled, logError);
if (!callback) return undefined;
return (...args: Args) => {
try {
callback(...args);
} catch (e) {
console.error('[vjs-queue]', e);
}
};
};
this.#onDispatch = safeCallback(config.onDispatch);
this.#onSettled = safeCallback(config.onSettled);
} }
get queued(): Readonly<PublicQueuedRecord<Tasks>> { get queued(): Readonly<PublicQueuedRecord<Tasks>> {
return Object.freeze({ ...this.#queued }); return Object.freeze({ ...this.#queued });
} }
get pending(): Readonly<PendingRecord<Tasks>> { /**
return Object.freeze({ ...this.#pending }); * Map of task key -> task (pending, success, or error).
*/
get tasks(): Readonly<TasksRecord<Tasks>> {
return Object.freeze({ ...this.#tasks });
} }
get destroyed(): boolean { get destroyed(): boolean {
@@ -208,7 +241,7 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
* Check if a task with the given key is currently pending (executing). * Check if a task with the given key is currently pending (executing).
*/ */
isPending(key: keyof Tasks): boolean { isPending(key: keyof Tasks): boolean {
return key in this.#pending; return this.#tasks[key]?.status === 'pending';
} }
/** /**
@@ -219,11 +252,53 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
} }
/** /**
* Subscribe to pending state changes. * Check if a task with the given key is settled (success or error).
*/
isSettled(key: keyof Tasks): boolean {
const task = this.#tasks[key];
return task?.status === 'success' || task?.status === 'error';
}
/**
* Clear settled task(s).
* *
* Fires when tasks are dispatched or settled. * - If key provided: clears that specific settled task (no-op if pending or doesn't exist)
* - If no key: clears all settled tasks (pending tasks are preserved)
* *
* @param listener - Callback receiving the current pending map * @param key - Optional task key to reset. If omitted, resets all settled tasks.
*/
reset(key?: keyof Tasks): void {
if (!isUndefined(key)) {
const task = this.#tasks[key];
if (!task || task.status === 'pending') return;
delete this.#tasks[key];
this.#notifySubscribers();
return;
}
// Reset all settled tasks
let cleared = false;
for (const key of Reflect.ownKeys(this.#tasks)) {
const task = this.#tasks[key];
if (task && task.status !== 'pending') {
delete this.#tasks[key];
cleared = true;
}
}
if (cleared) {
this.#notifySubscribers();
}
}
/**
* Subscribe to task state changes.
*
* Fires when tasks are dispatched, settled, or reset.
*
* @param listener - Callback receiving the current tasks map
* @returns Unsubscribe function * @returns Unsubscribe function
*/ */
subscribe(listener: QueueListener<Tasks>): () => void { subscribe(listener: QueueListener<Tasks>): () => void {
@@ -236,7 +311,7 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
#notifySubscribers(): void { #notifySubscribers(): void {
if (this.#subscribers.size === 0) return; if (this.#subscribers.size === 0) return;
const snapshot = this.pending; const snapshot = this.tasks;
for (const listener of this.#subscribers) { for (const listener of this.#subscribers) {
try { try {
listener(snapshot); listener(snapshot);
@@ -262,7 +337,13 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
delete this.#queued[key]; delete this.#queued[key];
// Abort any pending task with same key // Abort any pending task with same key
this.#pending[key]?.abort.abort(new StoreError('SUPERSEDED')); const existing = this.#tasks[key];
if (existing?.status === 'pending') {
existing.abort.abort(new StoreError('SUPERSEDED'));
}
// Clear any settled task for this key (new request replaces it)
delete this.#tasks[key];
return new Promise<Tasks[K]['output']>((resolve, reject) => { return new Promise<Tasks[K]['output']>((resolve, reject) => {
const task: QueuedTask = { const task: QueuedTask = {
@@ -307,24 +388,37 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
}); });
} }
dequeue<K extends keyof Tasks>(key: K): boolean { /**
const queued = this.#queued[key]; * Cancel queued task(s) waiting to execute.
if (!queued) return false; *
* - If key provided: cancels that specific queued task
* - If no key: cancels all queued tasks
*
* @param key - Optional task key to cancel
* @returns true if any task was cancelled
*/
cancel(key?: keyof Tasks): boolean {
if (!isUndefined(key)) {
const queued = this.#queued[key];
if (!queued) return false;
queued.invalidate?.(); queued.invalidate?.();
queued.reject(new StoreError('REMOVED')); queued.reject(new StoreError('REMOVED'));
delete this.#queued[key]; delete this.#queued[key];
return true; return true;
} }
clear(): void { // Cancel all queued
const hadQueued = Object.keys(this.#queued).length > 0;
for (const queued of Object.values(this.#queued)) { for (const queued of Object.values(this.#queued)) {
queued.invalidate?.(); queued.invalidate?.();
queued.reject(new StoreError('REMOVED')); queued.reject(new StoreError('REMOVED'));
} }
this.#queued = {}; this.#queued = {};
return hadQueued;
} }
async flush(key?: keyof Tasks): Promise<void> { async flush(key?: keyof Tasks): Promise<void> {
@@ -338,18 +432,32 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
await Promise.allSettled(keys.map(k => this.#flushKey(k))); await Promise.allSettled(keys.map(k => this.#flushKey(k)));
} }
abort<K extends keyof Tasks>(key: K): void { /**
// Reject queued * Abort task(s) - both queued (waiting) and pending (executing).
const queued = this.#queued[key]; *
queued?.invalidate?.(); * - If key provided: aborts that specific task
queued?.reject(new StoreError('ABORTED')); * - If no key: aborts all tasks
delete this.#queued[key]; *
* @param key - Optional task key to abort
*/
abort(key?: keyof Tasks): void {
if (!isUndefined(key)) {
// Reject queued
const queued = this.#queued[key];
queued?.invalidate?.();
queued?.reject(new StoreError('ABORTED'));
delete this.#queued[key];
// Abort pending // Abort pending task
this.#pending[key]?.abort.abort(new StoreError('ABORTED')); const task = this.#tasks[key];
} if (task?.status === 'pending') {
task.abort.abort(new StoreError('ABORTED'));
}
abortAll(): void { return;
}
// Abort all
const error = new StoreError('ABORTED'); const error = new StoreError('ABORTED');
// Reject all queued // Reject all queued
@@ -360,9 +468,11 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
this.#queued = {}; this.#queued = {};
// Abort all pending // Abort all pending tasks
for (const pending of Object.values(this.#pending)) { for (const task of Object.values(this.#tasks)) {
pending.abort.abort(error); if (task?.status === 'pending') {
task.abort.abort(error);
}
} }
} }
@@ -370,8 +480,9 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
if (this.#destroyed) return; if (this.#destroyed) return;
this.#destroyed = true; this.#destroyed = true;
this.abortAll(); this.abort();
this.#subscribers.clear(); this.#subscribers.clear();
this.#tasks = {};
} }
async #flushKey(key: keyof Tasks): Promise<void> { async #flushKey(key: keyof Tasks): Promise<void> {
@@ -393,7 +504,8 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
const abort = new AbortController(); const abort = new AbortController();
const startedAt = Date.now(); const startedAt = Date.now();
const pending: PendingTask = { const pendingTask: PendingTask = {
status: 'pending',
id, id,
name, name,
key, key,
@@ -403,9 +515,9 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
meta, meta,
}; };
this.#pending[key as keyof Tasks] = pending; this.#tasks[key as keyof Tasks] = pendingTask;
this.#notifySubscribers(); this.#notifySubscribers();
this.#onDispatch?.(pending); this.#onDispatch?.(pendingTask);
try { try {
if (abort.signal.aborted) { if (abort.signal.aborted) {
@@ -420,28 +532,38 @@ export class Queue<Tasks extends TaskRecord = DefaultTaskRecord> {
resolve(result); resolve(result);
this.#onSettled?.(pending, { const successTask: SuccessTask = {
...pendingTask,
status: 'success', status: 'success',
duration: Date.now() - startedAt, settledAt: Date.now(),
output: result, output: result,
}); };
// Only update if we're still the current task for this key
if (this.#tasks[key as keyof Tasks] === pendingTask) {
this.#tasks[key as keyof Tasks] = successTask;
this.#notifySubscribers();
}
this.#onSettled?.(successTask);
} catch (error) { } catch (error) {
reject(error); reject(error);
const cancelled = abort.signal.aborted; const errorTask: ErrorTask = {
...pendingTask,
this.#onSettled?.(pending, { status: 'error',
status: cancelled ? 'cancelled' : 'error', settledAt: Date.now(),
duration: Date.now() - startedAt,
error, error,
}); cancelled: abort.signal.aborted,
} finally { };
const currentPending = this.#pending[key as keyof Tasks];
// Only remove if we're still the pending task for this key // Only update if we're still the current task for this key
if (currentPending === pending) { if (this.#tasks[key as keyof Tasks] === pendingTask) {
delete this.#pending[key]; this.#tasks[key as keyof Tasks] = errorTask;
this.#notifySubscribers(); this.#notifySubscribers();
} }
this.#onSettled?.(errorTask);
} }
} }
} }
packages/store/src/core/store.ts
+5 -4
View File
@@ -1,4 +1,4 @@
import type { PendingTask, TaskContext } from './queue'; import type { PendingTask, Task, TaskContext } from './queue';
import type { RequestMeta, RequestMetaInit, ResolvedRequestConfig } from './request'; import type { RequestMeta, RequestMetaInit, ResolvedRequestConfig } from './request';
import type { AnySlice, InferSliceTarget, Slice, UnionSliceRequests, UnionSliceState, UnionSliceTasks } from './slice'; import type { AnySlice, InferSliceTarget, Slice, UnionSliceRequests, UnionSliceState, UnionSliceTasks } from './slice';
import type { StateFactory } from './state'; import type { StateFactory } from './state';
@@ -138,7 +138,7 @@ export class Store<Target, Slices extends AnySlice<Target>[] = AnySlice<Target>[
this.#attachAbort?.abort(); this.#attachAbort?.abort();
this.#attachAbort = null; this.#attachAbort = null;
this.#target = null; this.#target = null;
this.#queue.abortAll(); this.#queue.abort();
this.#resetState(); this.#resetState();
} }
@@ -318,10 +318,11 @@ export class Store<Target, Slices extends AnySlice<Target>[] = AnySlice<Target>[
handler, handler,
}); });
} catch (error) { } catch (error) {
const pending = this.#queue.pending as Record<string | symbol, PendingTask | undefined>; const tasks = this.#queue.tasks as Record<string | symbol, Task | undefined>;
const task = tasks[key];
this.#handleError({ this.#handleError({
request: pending[key], request: task?.status === 'pending' ? task : undefined,
error, error,
}); });
packages/store/src/core/tests/queue.test.ts
+434 -30
View File
@@ -163,8 +163,8 @@ describe('queue', () => {
}); });
}); });
describe('dequeue', () => { describe('cancel', () => {
it('removes queued task', async () => { it('cancel(key) removes specific queued task', async () => {
const queue = createQueue({ const queue = createQueue({
scheduler: delay(100), scheduler: delay(100),
}); });
@@ -172,29 +172,33 @@ describe('queue', () => {
const handler = vi.fn(); const handler = vi.fn();
const promise = queue.enqueue({ name: 'test', key: 'k', handler }); const promise = queue.enqueue({ name: 'test', key: 'k', handler });
expect(queue.dequeue('k')).toBe(true); expect(queue.cancel('k')).toBe(true);
expect(queue.dequeue('k')).toBe(false); expect(queue.cancel('k')).toBe(false);
vi.advanceTimersByTime(100); vi.advanceTimersByTime(100);
await expect(promise).rejects.toMatchObject({ code: 'REMOVED' }); await expect(promise).rejects.toMatchObject({ code: 'REMOVED' });
expect(handler).not.toHaveBeenCalled(); expect(handler).not.toHaveBeenCalled();
}); });
});
describe('clear', () => { it('cancel() clears all queued tasks', async () => {
it('clears all queued tasks', async () => {
const queue = createQueue({ scheduler: delay(100) }); const queue = createQueue({ scheduler: delay(100) });
const p1 = queue.enqueue({ name: 'a', key: 'a', handler: vi.fn() }); const p1 = queue.enqueue({ name: 'a', key: 'a', handler: vi.fn() });
const p2 = queue.enqueue({ name: 'b', key: 'b', handler: vi.fn() }); const p2 = queue.enqueue({ name: 'b', key: 'b', handler: vi.fn() });
queue.clear(); expect(queue.cancel()).toBe(true);
vi.advanceTimersByTime(100); vi.advanceTimersByTime(100);
await expect(p1).rejects.toMatchObject({ code: 'REMOVED' }); await expect(p1).rejects.toMatchObject({ code: 'REMOVED' });
await expect(p2).rejects.toMatchObject({ code: 'REMOVED' }); await expect(p2).rejects.toMatchObject({ code: 'REMOVED' });
expect(Reflect.ownKeys(queue.queued).length).toBe(0); expect(Reflect.ownKeys(queue.queued).length).toBe(0);
}); });
it('cancel() returns false when no queued tasks', () => {
const queue = createQueue();
expect(queue.cancel()).toBe(false);
});
}); });
describe('flush', () => { describe('flush', () => {
@@ -280,7 +284,7 @@ describe('queue', () => {
); );
}); });
it('onSettled called with success', async () => { it('onSettled called with success task', async () => {
vi.useRealTimers(); vi.useRealTimers();
const onSettled = vi.fn(); const onSettled = vi.fn();
@@ -293,12 +297,11 @@ describe('queue', () => {
}); });
expect(onSettled).toHaveBeenCalledWith( expect(onSettled).toHaveBeenCalledWith(
expect.objectContaining({ name: 'task' }), expect.objectContaining({ name: 'task', status: 'success', output: 'done' }),
expect.objectContaining({ status: 'success' }),
); );
}); });
it('onSettled called with error status', async () => { it('onSettled called with error task', async () => {
vi.useRealTimers(); vi.useRealTimers();
const onSettled = vi.fn(); const onSettled = vi.fn();
@@ -313,10 +316,10 @@ describe('queue', () => {
}); });
await expect(promise).rejects.toThrow('oops'); await expect(promise).rejects.toThrow('oops');
expect(onSettled).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ status: 'error' })); expect(onSettled).toHaveBeenCalledWith(expect.objectContaining({ status: 'error', cancelled: false }));
}); });
it('onSettled called with cancelled status', async () => { it('onSettled called with cancelled task', async () => {
vi.useRealTimers(); vi.useRealTimers();
const onSettled = vi.fn(); const onSettled = vi.fn();
@@ -339,8 +342,7 @@ describe('queue', () => {
await promise.catch(() => {}); await promise.catch(() => {});
expect(onSettled).toHaveBeenCalledWith( expect(onSettled).toHaveBeenCalledWith(
expect.objectContaining({ name: 'first' }), expect.objectContaining({ name: 'first', status: 'error', cancelled: true }),
expect.objectContaining({ status: 'cancelled' }),
); );
}); });
}); });
@@ -436,7 +438,7 @@ describe('queue', () => {
// Wait for task to start // Wait for task to start
await new Promise(r => setTimeout(r, 10)); await new Promise(r => setTimeout(r, 10));
expect(Reflect.ownKeys(queue.pending).length).toBe(1); expect(queue.tasks.k?.status).toBe('pending');
// Destroy queue // Destroy queue
queue.destroy(); queue.destroy();
@@ -446,8 +448,8 @@ describe('queue', () => {
expect(cleanupSpy).toHaveBeenCalledWith('aborted'); expect(cleanupSpy).toHaveBeenCalledWith('aborted');
expect(cleanupSpy).toHaveBeenCalledWith('cleanup'); expect(cleanupSpy).toHaveBeenCalledWith('cleanup');
// Pending object should be empty (self-cleaned) // After destroy, all tasks are cleared for memory cleanup
expect(Reflect.ownKeys(queue.pending).length).toBe(0); expect(queue.tasks.k).toBeUndefined();
}); });
it('handles scheduler error without double-cleanup when already flushed', async () => { it('handles scheduler error without double-cleanup when already flushed', async () => {
@@ -575,7 +577,7 @@ describe('queue', () => {
expect(listener).toHaveBeenCalledTimes(2); expect(listener).toHaveBeenCalledTimes(2);
}); });
it('notifies with pending map when task dispatches', async () => { it('notifies with tasks map on dispatch and settlement', async () => {
vi.useRealTimers(); vi.useRealTimers();
const queue = createQueue(); const queue = createQueue();
@@ -602,18 +604,19 @@ describe('queue', () => {
// First call should have pending task // First call should have pending task
expect(listener).toHaveBeenCalledTimes(1); expect(listener).toHaveBeenCalledTimes(1);
const pendingObj = listener.mock.calls[0]![0] as Record<string, unknown>; const pendingSnapshot = listener.mock.calls[0]![0] as Record<string, { status: string }>;
expect(Reflect.ownKeys(pendingObj).length).toBe(1); expect(Reflect.ownKeys(pendingSnapshot).length).toBe(1);
expect('test-key' in pendingObj).toBe(true); expect(pendingSnapshot['test-key']?.status).toBe('pending');
// Complete the handler // Complete the handler
resolveHandler!(); resolveHandler!();
await promise; await promise;
// Second call should have empty pending // Second call should have settled task (success)
expect(listener).toHaveBeenCalledTimes(2); expect(listener).toHaveBeenCalledTimes(2);
const settledObj = listener.mock.calls[1]![0] as Record<string, unknown>; const settledSnapshot = listener.mock.calls[1]![0] as Record<string, { status: string }>;
expect(Reflect.ownKeys(settledObj).length).toBe(0); expect(Reflect.ownKeys(settledSnapshot).length).toBe(1);
expect(settledSnapshot['test-key']?.status).toBe('success');
}); });
it('unsubscribe stops notifications', async () => { it('unsubscribe stops notifications', async () => {
@@ -683,15 +686,15 @@ describe('queue', () => {
consoleSpy.mockRestore(); consoleSpy.mockRestore();
}); });
it('provides strongly typed pending object', async () => { it('provides strongly typed tasks object', async () => {
vi.useRealTimers(); vi.useRealTimers();
// Use default queue - type safety is validated at compile time // Use default queue - type safety is validated at compile time
const queue = createQueue(); const queue = createQueue();
queue.subscribe((pending) => { queue.subscribe((tasks) => {
// Pending is a frozen object // Tasks is a frozen object
const task = pending.playback; const task = tasks.playback;
if (task) { if (task) {
expect(task.key).toBe('playback'); expect(task.key).toBe('playback');
expect(task.name).toBeDefined(); expect(task.name).toBeDefined();
@@ -705,5 +708,406 @@ describe('queue', () => {
}); });
}); });
}); });
describe('task lifecycle', () => {
it('task starts as pending and transitions to success', async () => {
vi.useRealTimers();
const queue = createQueue();
const promise = queue.enqueue({
name: 'task',
key: 'k',
handler: async () => {
await new Promise(r => setTimeout(r, 10));
return 'result';
},
});
// Task should be pending
await new Promise(r => setTimeout(r, 5));
const pendingTask = queue.tasks.k;
expect(pendingTask?.status).toBe('pending');
expect(pendingTask?.name).toBe('task');
// Wait for completion
await promise;
// Task should be success
const successTask = queue.tasks.k;
expect(successTask?.status).toBe('success');
if (successTask?.status === 'success') {
expect(successTask.output).toBe('result');
expect(successTask.settledAt).toBeGreaterThan(successTask.startedAt);
}
});
it('task starts as pending and transitions to error', async () => {
vi.useRealTimers();
const queue = createQueue();
const error = new Error('test error');
const promise = queue.enqueue({
name: 'task',
key: 'k',
handler: async () => {
await new Promise(r => setTimeout(r, 10));
throw error;
},
});
// Task should be pending
await new Promise(r => setTimeout(r, 5));
expect(queue.tasks.k?.status).toBe('pending');
// Wait for failure
await expect(promise).rejects.toThrow('test error');
// Task should be error
const errorTask = queue.tasks.k;
expect(errorTask?.status).toBe('error');
if (errorTask?.status === 'error') {
expect(errorTask.error).toBe(error);
expect(errorTask.cancelled).toBe(false);
}
});
it('aborted task has cancelled flag set to true', async () => {
vi.useRealTimers();
const queue = createQueue();
const promise = queue.enqueue({
name: 'task',
key: 'k',
handler: async ({ signal }) => {
await new Promise((_, reject) => {
signal.addEventListener('abort', () => reject(signal.reason));
setTimeout(() => {}, 1000);
});
},
});
// Wait for task to start
await new Promise(r => setTimeout(r, 10));
expect(queue.tasks.k?.status).toBe('pending');
// Abort the task
queue.abort('k');
await promise.catch(() => {});
// Task should be error with cancelled=true
const errorTask = queue.tasks.k;
expect(errorTask?.status).toBe('error');
if (errorTask?.status === 'error') {
expect(errorTask.cancelled).toBe(true);
}
});
it('new request replaces settled task', async () => {
vi.useRealTimers();
const queue = createQueue();
// First request
await queue.enqueue({
name: 'first',
key: 'k',
handler: async () => 'first-result',
});
expect(queue.tasks.k?.status).toBe('success');
if (queue.tasks.k?.status === 'success') {
expect(queue.tasks.k.output).toBe('first-result');
}
// Second request replaces settled task
await queue.enqueue({
name: 'second',
key: 'k',
handler: async () => 'second-result',
});
expect(queue.tasks.k?.status).toBe('success');
if (queue.tasks.k?.status === 'success') {
expect(queue.tasks.k.output).toBe('second-result');
expect(queue.tasks.k.name).toBe('second');
}
});
});
describe('reset', () => {
it('clears settled task', async () => {
vi.useRealTimers();
const queue = createQueue();
await queue.enqueue({
name: 'task',
key: 'k',
handler: async () => 'result',
});
expect(queue.tasks.k?.status).toBe('success');
queue.reset('k');
expect(queue.tasks.k).toBeUndefined();
});
it('is no-op when task is pending', async () => {
vi.useRealTimers();
const queue = createQueue();
const promise = queue.enqueue({
name: 'task',
key: 'k',
handler: async () => {
await new Promise(r => setTimeout(r, 50));
return 'result';
},
});
// Wait for task to start
await new Promise(r => setTimeout(r, 10));
expect(queue.tasks.k?.status).toBe('pending');
// Reset should be no-op
queue.reset('k');
expect(queue.tasks.k?.status).toBe('pending');
await promise;
});
it('is no-op when task does not exist', () => {
const queue = createQueue();
// Should not throw
queue.reset('nonexistent');
expect(queue.tasks.nonexistent).toBeUndefined();
});
it('notifies subscribers when reset clears a task', async () => {
vi.useRealTimers();
const queue = createQueue();
const listener = vi.fn();
await queue.enqueue({
name: 'task',
key: 'k',
handler: async () => 'result',
});
queue.subscribe(listener);
queue.reset('k');
expect(listener).toHaveBeenCalledTimes(1);
const snapshot = listener.mock.calls[0]![0] as Record<string, unknown>;
expect(snapshot.k).toBeUndefined();
});
it('does not notify subscribers when task does not exist', () => {
const queue = createQueue();
const listener = vi.fn();
queue.subscribe(listener);
queue.reset('nonexistent');
expect(listener).not.toHaveBeenCalled();
});
it('resets all settled tasks when no key provided', async () => {
vi.useRealTimers();
const queue = createQueue();
// Create multiple settled tasks
await queue.enqueue({ name: 'a', key: 'a', handler: async () => 'a-result' });
await queue.enqueue({ name: 'b', key: 'b', handler: async () => 'b-result' });
expect(queue.tasks.a?.status).toBe('success');
expect(queue.tasks.b?.status).toBe('success');
// Reset all
queue.reset();
expect(queue.tasks.a).toBeUndefined();
expect(queue.tasks.b).toBeUndefined();
});
it('preserves pending tasks when resetting all', async () => {
vi.useRealTimers();
const queue = createQueue();
// Create a settled task
await queue.enqueue({ name: 'settled', key: 'settled', handler: async () => 'done' });
// Create a pending task
const pendingPromise = queue.enqueue({
name: 'pending',
key: 'pending',
handler: async () => {
await new Promise(r => setTimeout(r, 100));
return 'pending-done';
},
});
await new Promise(r => setTimeout(r, 10));
expect(queue.tasks.settled?.status).toBe('success');
expect(queue.tasks.pending?.status).toBe('pending');
// Reset all - should only clear settled
queue.reset();
expect(queue.tasks.settled).toBeUndefined();
expect(queue.tasks.pending?.status).toBe('pending');
await pendingPromise;
});
});
describe('isSettled', () => {
it('returns true for success task', async () => {
vi.useRealTimers();
const queue = createQueue();
await queue.enqueue({ name: 'task', key: 'k', handler: async () => 'result' });
expect(queue.isSettled('k')).toBe(true);
});
it('returns true for error task', async () => {
vi.useRealTimers();
const queue = createQueue();
const promise = queue.enqueue({
name: 'task',
key: 'k',
handler: async () => {
throw new Error('fail');
},
});
await promise.catch(() => {});
expect(queue.isSettled('k')).toBe(true);
});
it('returns false for pending task', async () => {
vi.useRealTimers();
const queue = createQueue();
const promise = queue.enqueue({
name: 'task',
key: 'k',
handler: async () => {
await new Promise(r => setTimeout(r, 50));
return 'result';
},
});
await new Promise(r => setTimeout(r, 10));
expect(queue.isSettled('k')).toBe(false);
await promise;
});
it('returns false for non-existent task', () => {
const queue = createQueue();
expect(queue.isSettled('nonexistent')).toBe(false);
});
});
describe('destroy cleanup', () => {
it('clears all task references on destroy', async () => {
vi.useRealTimers();
const queue = createQueue();
await queue.enqueue({ name: 'task', key: 'k', handler: async () => 'result' });
expect(queue.tasks.k?.status).toBe('success');
queue.destroy();
expect(Reflect.ownKeys(queue.tasks).length).toBe(0);
});
});
describe('tasks getter', () => {
it('returns frozen snapshot', async () => {
vi.useRealTimers();
const queue = createQueue();
await queue.enqueue({ name: 'task', key: 'k', handler: async () => 'result' });
const tasks = queue.tasks;
expect(Object.isFrozen(tasks)).toBe(true);
});
it('returns independent snapshots', async () => {
vi.useRealTimers();
const queue = createQueue();
await queue.enqueue({ name: 'first', key: 'k', handler: async () => 'first' });
const snapshot1 = queue.tasks;
await queue.enqueue({ name: 'second', key: 'k', handler: async () => 'second' });
const snapshot2 = queue.tasks;
// Snapshots should be independent
expect(snapshot1).not.toBe(snapshot2);
if (snapshot1.k?.status === 'success' && snapshot2.k?.status === 'success') {
expect(snapshot1.k.output).toBe('first');
expect(snapshot2.k.output).toBe('second');
}
});
});
describe('symbol keys', () => {
it('supports symbol keys', async () => {
vi.useRealTimers();
const queue = createQueue();
const key = Symbol('task');
await queue.enqueue({
name: 'task',
key,
handler: async () => 'result',
});
expect(queue.tasks[key]?.status).toBe('success');
if (queue.tasks[key]?.status === 'success') {
expect(queue.tasks[key].output).toBe('result');
}
});
});
describe('meta propagation', () => {
it('meta defaults to null when not provided', async () => {
vi.useRealTimers();
const queue = createQueue();
await queue.enqueue({
name: 'task',
key: 'k',
handler: async () => 'result',
});
expect(queue.tasks.k?.meta).toBeNull();
});
});
}); });
}); });
packages/utils/src/function/index.ts
+1
View File
@@ -1 +1,2 @@
export { composeCallbacks } from './compose-callbacks'; export { composeCallbacks } from './compose-callbacks';
export { tryCatch } from './try-catch';
packages/utils/src/function/tests/try-catch.test.ts
+79
View File
@@ -0,0 +1,79 @@
import { describe, expect, it, vi } from 'vitest';
import { tryCatch } from '../try-catch';
describe('tryCatch', () => {
it('returns undefined if fn is undefined', () => {
expect(tryCatch(undefined)).toBeUndefined();
});
it('returns undefined if fn is null', () => {
// @ts-expect-error - testing null input
expect(tryCatch(null)).toBeUndefined();
});
it('calls the wrapped function with arguments', () => {
const fn = vi.fn((a: number, b: number) => a + b);
const wrapped = tryCatch(fn);
const result = wrapped?.(1, 2);
expect(fn).toHaveBeenCalledWith(1, 2);
expect(result).toBe(3);
});
it('returns the function result when no error', () => {
const fn = () => 'result';
const wrapped = tryCatch(fn);
expect(wrapped?.()).toBe('result');
});
it('catches errors and calls onError', () => {
const error = new Error('test error');
const fn = () => {
throw error;
};
const onError = vi.fn();
const wrapped = tryCatch(fn, onError);
const result = wrapped?.();
expect(onError).toHaveBeenCalledWith(error);
expect(result).toBeUndefined();
});
it('uses console.error as default onError', () => {
const consoleSpy = vi.spyOn(console, 'error').mockImplementation(() => {});
const error = new Error('test error');
const fn = () => {
throw error;
};
const wrapped = tryCatch(fn);
wrapped?.();
expect(consoleSpy).toHaveBeenCalledWith(error);
consoleSpy.mockRestore();
});
it('does not throw when wrapped function throws', () => {
const fn = () => {
throw new Error('should not propagate');
};
const onError = vi.fn();
const wrapped = tryCatch(fn, onError);
expect(() => wrapped?.()).not.toThrow();
});
it('preserves function type signature', () => {
const fn = (name: string, age: number): string => `${name} is ${age}`;
const wrapped = tryCatch(fn);
// TypeScript should infer correct types
const result: string | undefined = wrapped?.('Alice', 30);
expect(result).toBe('Alice is 30');
});
});
packages/utils/src/function/try-catch.ts
+28
View File
@@ -0,0 +1,28 @@
/**
* Wrap a function to catch and handle errors instead of throwing.
*
* @param fn - Function to wrap (can be undefined)
* @param onError - Error handler (defaults to console.error)
* @returns Wrapped function that never throws, or undefined if fn is undefined
*
* @example
* ```ts
* const safeFn = tryCatch(riskyFn, (e) => logger.error(e));
* safeFn?.(); // Never throws
* ```
*/
export function tryCatch<T extends (...args: any[]) => unknown>(
fn: T | undefined,
onError: (error: unknown) => void = console.error,
): T | undefined {
if (!fn) return undefined;
return ((...args: Parameters<T>) => {
try {
return fn(...args);
} catch (error) {
onError(error);
return undefined;
}
}) as T;
}