Merge pull request #669 from dahlia/docfix

Fix and clarify Cloudflare Workers documentation

Author: dahlia

docs/manual/deploy.md

@@ -163,7 +163,7 @@ on your infrastructure:
 
 Development
 :   Use [`MemoryKvStore`](./kv.md#memorykvstore) and
-    [`InProcessMessageQueue`](./mq.md#inprocessmessagequeu) for quick setup.
+    [`InProcessMessageQueue`](./mq.md#inprocessmessagequeue) for quick setup.
 
 Production
 :   Consider [`PostgresKvStore`](./kv.md#postgreskvstore) and
@@ -291,8 +291,8 @@ export default {
     for (const message of batch.messages) {
       try {
         await federation.processQueuedTask(
-          message.body as unknown as Message,
           env,
+          message.body as unknown as Message,
         );
         message.ack();
       } catch (error) {
@@ -303,6 +303,13 @@ export default {
 };
 ~~~~
 
+If you use queue ordering keys on Cloudflare Workers, instantiate
+`WorkersMessageQueue` with an `orderingKv` namespace and call
+`WorkersMessageQueue.processMessage()` before
+`Federation.processQueuedTask()`.  See the
+[*`WorkersMessageQueue`* section](./mq.md#workersmessagequeue-cloudflare-workers-only)
+for a complete example and caveats about best-effort ordering.
+
 ### Example deployment
 
 For a complete working example, see the [Cloudflare Workers example]

docs/manual/mq.md

@@ -604,6 +604,56 @@ export default {
 > process the messages.  The `queue()` method is the only way to consume
 > messages from the queue in Cloudflare Workers.
 
+> [!NOTE]
+> If you use `~MessageQueueEnqueueOptions.orderingKey` with
+> `WorkersMessageQueue`, you also need to provide a KV namespace for ordering
+> locks and pass each raw queue message through
+> `~WorkersMessageQueue.processMessage()` before calling
+> `Federation.processQueuedTask()`.  Otherwise, the ordering key is embedded in
+> the message, but not enforced when the worker consumes it.
+>
+> ~~~~ typescript
+> import { createFederationBuilder, type Message } from "@fedify/fedify";
+> import { WorkersKvStore, WorkersMessageQueue } from "@fedify/cfworkers";
+>
+> type Env = {
+>   KV_NAMESPACE: KVNamespace<string>;
+>   QUEUE_BINDING: Queue;
+>   ORDERING_KV: KVNamespace<string>;
+> };
+>
+> const builder = createFederationBuilder<Env>();
+>
+> export default {
+>   async queue(batch: MessageBatch<unknown>, env: Env): Promise<void> {
+>     const queue = new WorkersMessageQueue(env.QUEUE_BINDING, {
+>       orderingKv: env.ORDERING_KV,
+>     });
+>     const federation = await builder.build({
+>       kv: new WorkersKvStore(env.KV_NAMESPACE),
+>       queue,
+>     });
+>
+>     for (const message of batch.messages) {
+>       const result = await queue.processMessage(message.body);
+>       if (!result.shouldProcess) {
+>         message.retry();
+>         continue;
+>       }
+>       try {
+>         await federation.processQueuedTask(
+>           env,
+>           result.message as Message,
+>         );
+>         message.ack();
+>       } finally {
+>         await result.release?.();
+>       }
+>     }
+>   },
+> };
+> ~~~~
+
 [Cloudflare Workers]: https://workers.cloudflare.com/
 [Cloudflare Queues]: https://developers.cloudflare.com/queues/
 

packages/cfworkers/README.md

@@ -16,7 +16,7 @@ implementations for [Cloudflare Workers]:
  -  [`WorkersMessageQueue`]
 
 ~~~~ typescript
-import type { Federation } from "@fedify/fedify";
+import { createFederation, type Message } from "@fedify/fedify";
 import { WorkersKvStore, WorkersMessageQueue } from "@fedify/cfworkers";
 
 export default {
@@ -26,22 +26,25 @@ export default {
       queue: new WorkersMessageQueue(env.QUEUE_BINDING),
       // ... other options
     });
-    
-    return federation.handle(request, { contextData: env });
+
+    return federation.fetch(request, { contextData: env });
   },
-  
+
   async queue(batch, env, ctx) {
     const federation = createFederation({
       kv: new WorkersKvStore(env.KV_BINDING),
       queue: new WorkersMessageQueue(env.QUEUE_BINDING),
       // ... other options
     });
-    
+
     for (const message of batch.messages) {
-      await federation.processQueuedTask(message.body);
+      await federation.processQueuedTask(
+        env,
+        message.body as unknown as Message,
+      );
     }
   }
-} satisfies ExportedHandler<{ 
+} satisfies ExportedHandler<{
   KV_BINDING: KVNamespace<string>;
   QUEUE_BINDING: Queue;
 }>;
@@ -101,6 +104,14 @@ management.
 > process the messages.  The `queue()` method is the only way to consume
 > messages from the queue in Cloudflare Workers.
 
+> [!NOTE]
+> If you use `orderingKey` when enqueueing messages, construct
+> `WorkersMessageQueue` with an `orderingKv` namespace and pass each raw queue
+> message through `WorkersMessageQueue.processMessage()` before calling
+> `Federation.processQueuedTask()`.  This acquires and releases the best-effort
+> ordering lock for that key.  You can also customize the lock behavior with
+> the `orderingKeyPrefix` and `orderingLockTtl` options.
+
 [Cloudflare Queues]: https://developers.cloudflare.com/queues/