From 2fc43bcccb4cdd8d99c28d5e2fa3d07141330621 Mon Sep 17 00:00:00 2001 From: Igor Savin Date: Fri, 12 Jun 2026 13:15:05 +0300 Subject: [PATCH] docs(sqs): make sync/async resolver support explicit Invalidation trigger resolvers already support both synchronous and asynchronous forms: the `InvalidationResolver` type returns `ResolverOutput | Promise>`, and the dispatch pipeline `await`s the result, so a non-Promise return resolves immediately and a Promise is awaited before actions are applied. Make this explicit: - Expand the `InvalidationResolver` JSDoc to spell out that it may be sync or async. - Flesh out the README "Resolver semantics" section with sync and async examples instead of the one-line aside. - Add an end-to-end test exercising an async resolver that derives keys via an awaited I/O lookup. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/sqs/README.md | 13 ++- packages/sqs/lib/triggers/types.ts | 9 +- .../sqs/test/SqsInvalidationTrigger.spec.ts | 84 +++++++++++++++++++ 3 files changed, 103 insertions(+), 3 deletions(-) diff --git a/packages/sqs/README.md b/packages/sqs/README.md index 31ee959..35b3c28 100644 --- a/packages/sqs/README.md +++ b/packages/sqs/README.md @@ -551,7 +551,18 @@ type GroupInvalidationAction = | { kind: 'clear' } ``` -Resolvers may be `async`; the trigger awaits before applying actions. +A resolver may be **synchronous or asynchronous** — both are fully supported. Return the action(s) directly for a simple field-to-key mapping, or return a `Promise` (e.g. an `async` function) when deriving the affected keys requires I/O such as a database or service lookup. The trigger always `await`s the resolver before applying actions, so the two forms behave identically: + +```ts +// Synchronous: map a field straight to a key. +resolver: (msg) => ({ kind: 'delete', key: msg.userId }) + +// Asynchronous: look up the affected keys first. +resolver: async (msg) => { + const memberIds = await membership.listMemberIds(msg.teamId) + return { kind: 'deleteMany', keys: memberIds } +} +``` ### Error handling and retries diff --git a/packages/sqs/lib/triggers/types.ts b/packages/sqs/lib/triggers/types.ts index 4fbd354..9e07cd2 100644 --- a/packages/sqs/lib/triggers/types.ts +++ b/packages/sqs/lib/triggers/types.ts @@ -25,8 +25,13 @@ export type GroupInvalidationAction = export type ResolverOutput = TAction | readonly TAction[] | null | undefined /** - * Pure function turning a parsed upstream message into invalidation - * action(s). Return `undefined`/`null` to skip the message. + * Function turning a parsed upstream message into invalidation action(s). + * Return `undefined`/`null` to skip the message. + * + * May be **synchronous or asynchronous**: return the action(s) directly, or a + * `Promise` resolving to them (e.g. when you need to look up the affected keys + * from a database or another service). The trigger always `await`s the result + * before applying actions, so both forms are handled uniformly. */ export type InvalidationResolver = ( message: TMessage, diff --git a/packages/sqs/test/SqsInvalidationTrigger.spec.ts b/packages/sqs/test/SqsInvalidationTrigger.spec.ts index 19be3fe..02bf1b6 100644 --- a/packages/sqs/test/SqsInvalidationTrigger.spec.ts +++ b/packages/sqs/test/SqsInvalidationTrigger.spec.ts @@ -721,6 +721,90 @@ describe('SqsQueueInvalidationTrigger', () => { } }) + it('awaits an async resolver that derives keys via an I/O lookup', async () => { + const suffix = Math.random().toString(36).slice(2, 8) + const invalidationTopic = `async-src-${suffix}` + + const peer = createNotificationPair({ + publisher: { + dependencies: buildPublisherDeps(clients), + creationConfig: { topic: { Name: invalidationTopic } }, + }, + consumer: { + dependencies: buildConsumerDeps(clients), + creationConfig: { + topic: { Name: invalidationTopic }, + queue: { QueueName: `async-src-peer-q-${suffix}` }, + }, + }, + }) + + const loader = new Loader({ + inMemoryCache: IN_MEMORY_CACHE_CONFIG, + asyncCache: new StubAsyncCache('value'), + notificationConsumer: peer.consumer, + notificationPublisher: peer.publisher, + }) + + const upstreamQueueName = `domain-events-async-q-${suffix}` + await clients.sqsClient.send(new CreateQueueCommand({ QueueName: upstreamQueueName })) + const queueUrl = ( + await clients.sqsClient.send(new GetQueueUrlCommand({ QueueName: upstreamQueueName })) + ).QueueUrl! + + // Stand-in for a DB/service lookup the resolver must await before it knows + // which keys an event actually affects. + const membership: Record = { 'team-7': ['user-1', 'user-2'] } + const lookupMembers = (teamId: string) => + Promise.resolve(membership[teamId] ?? []) + + const trigger = new SqsQueueInvalidationTrigger({ + target: loader, + dependencies: buildConsumerDeps(clients), + sources: [ + { + locatorConfig: { queueUrl }, + bindings: [ + { + messageSchema: UPSTREAM_EVENT_SCHEMA, + resolver: async (m: UpstreamEvent) => { + if (m.eventType !== 'user.bulk-updated' || !m.userId) return null + const keys = await lookupMembers(m.userId) + return keys.length ? { kind: 'deleteMany', keys } : null + }, + }, + ], + }, + ], + }) + + try { + await loader.init() + await trigger.start() + + await loader.getAsyncOnly('user-1') + await loader.getAsyncOnly('user-2') + expect(loader.getInMemoryOnly('user-1')).toBe('value') + expect(loader.getInMemoryOnly('user-2')).toBe('value') + + // userId carries the team id this event fans out to. + await clients.sqsClient.send( + new SendMessageCommand({ + QueueUrl: queueUrl, + MessageBody: JSON.stringify({ eventType: 'user.bulk-updated', userId: 'team-7' }), + }), + ) + + await waitFor( + () => + loader.getInMemoryOnly('user-1') === undefined && + loader.getInMemoryOnly('user-2') === undefined, + ) + } finally { + await Promise.allSettled([trigger.stop(), peer.consumer.close(), peer.publisher.close()]) + } + }) + it('consumes from two SQS queues simultaneously', async () => { const suffix = Math.random().toString(36).slice(2, 8) const invalidationTopic = `multi-sqs-${suffix}`