feat(oidc): consume provider records atomically and report the winner

.changeset/oidc-atomic-consume.md

@@ -0,0 +1,5 @@
+---
+'seamless-auth-api': minor
+---
+
+`POST /oidc/payloads/:model/:id/consume` now answers `200 { consumed }` instead of `204` (`@seamless-auth/types` 0.34.0-next.4). It sets `consumedAt` in one conditional statement, and `consumed` is true only for the call that set it. Two concurrent refreshes with the same refresh token used to both pass the `consumedAt ??=` check and both rotate. Now exactly one wins, and the provider service treats the other as reuse. An expired record is never consumed.

openapi.json

@@ -12109,7 +12109,7 @@
     },
     "/oidc/payloads/{model}/{id}/consume": {
       "post": {
-        "summary": "Mark a provider record consumed",
+        "summary": "Mark a provider record consumed, and say whether this call did",
         "tags": ["OpenID Provider service"],
         "security": [{ "serviceToken": [] }],
         "parameters": [
@@ -12141,6 +12141,19 @@
           }
         ],
         "responses": {
+          "200": {
+            "description": "HTTP 200",
+            "content": {
+              "application/json": {
+                "example": { "consumed": true },
+                "schema": {
+                  "type": "object",
+                  "properties": { "consumed": { "type": "boolean" } },
+                  "required": ["consumed"]
+                }
+              }
+            }
+          },
           "400": {
             "description": "HTTP 400",
             "content": {

src/controllers/oidcEdge.ts

@@ -131,8 +131,8 @@ export async function getOidcPayloadByUid(req: Req<OidcPayloadUidParams>, res: R
 }
 
 export async function postOidcPayloadConsume(req: Req<OidcPayloadParams>, res: Response) {
-  await consumeOidcPayload(req.params.model, req.params.id);
-  return res.status(204).end();
+  const consumed = await consumeOidcPayload(req.params.model, req.params.id);
+  return res.status(200).json({ consumed });
 }
 
 export async function deleteOidcPayload(req: Req<OidcPayloadParams>, res: Response) {

src/generated/api.ts

@@ -11938,7 +11938,7 @@ export interface paths {
     };
     get?: never;
     put?: never;
-    /** Mark a provider record consumed */
+    /** Mark a provider record consumed, and say whether this call did */
     post: {
       parameters: {
         query?: never;
@@ -11961,6 +11961,22 @@ export interface paths {
       };
       requestBody?: never;
       responses: {
+        /** @description HTTP 200 */
+        200: {
+          headers: {
+            [name: string]: unknown;
+          };
+          content: {
+            /**
+             * @example {
+             *       "consumed": true
+             *     }
+             */
+            'application/json': {
+              consumed: boolean;
+            };
+          };
+        };
         /** @description HTTP 400 */
         400: {
           headers: {

src/routes/oidc.routes.ts

@@ -15,6 +15,7 @@ import {
   OidcEdgeConfigResponseSchema,
   OidcEdgeEventSchema,
   OidcJwksResponseSchema,
+  OidcPayloadConsumeResponseSchema,
   OidcPayloadGrantParamsSchema,
   OidcPayloadParamsSchema,
   OidcPayloadRecordSchema,
@@ -142,8 +143,11 @@ oidcRouter.post(
   '/payloads/:model/:id/consume',
   {
     ...edge,
-    summary: 'Mark a provider record consumed',
-    schemas: { params: OidcPayloadParamsSchema, response: guarded },
+    summary: 'Mark a provider record consumed, and say whether this call did',
+    schemas: {
+      params: OidcPayloadParamsSchema,
+      response: { 200: OidcPayloadConsumeResponseSchema, ...guarded },
+    },
   },
   postOidcPayloadConsume,
 );

src/services/oidcPayloadStore.ts

@@ -204,11 +204,24 @@ export function findOidcPayloadByUid(model: OidcPayloadModel, uid: string) {
   return findOne({ model, uid });
 }
 
+/**
+ * True only for the caller whose update set `consumedAt`. The condition is in the one
+ * statement, so of two refreshes racing with the same token exactly one wins, and the edge
+ * treats the other as reuse rather than issuing a second rotation.
+ */
 export async function consumeOidcPayload(model: OidcPayloadModel, id: string) {
-  await OidcPayload.update(
+  const [affected] = await OidcPayload.update(
     { consumedAt: new Date() },
-    { where: { model, id, consumedAt: { [Op.is]: null } } as WhereOptions<OidcPayloadAttributes> },
+    {
+      where: {
+        model,
+        id,
+        consumedAt: { [Op.is]: null },
+        expiresAt: { [Op.gt]: new Date() },
+      } as WhereOptions<OidcPayloadAttributes>,
+    },
   );
+  return affected > 0;
 }
 
 export async function destroyOidcPayload(model: OidcPayloadModel, id: string) {

tests/integration/oidc/oidcEdge.spec.ts

@@ -270,7 +270,13 @@ describe('/oidc/payloads', () => {
   });
 
   it('consumes, destroys and revokes by grant', async () => {
-    expect((await edge('post', '/oidc/payloads/AuthorizationCode/c1/consume')).status).toBe(204);
+    vi.mocked(consumeOidcPayload).mockResolvedValueOnce(true).mockResolvedValueOnce(false);
+    const won = await edge('post', '/oidc/payloads/AuthorizationCode/c1/consume');
+    expect(won.status).toBe(200);
+    expect(won.body).toEqual({ consumed: true });
+    expect((await edge('post', '/oidc/payloads/AuthorizationCode/c1/consume')).body).toEqual({
+      consumed: false,
+    });
     expect((await edge('delete', '/oidc/payloads/Session/s1')).status).toBe(204);
     expect((await edge('delete', '/oidc/payloads/RefreshToken/by-grant/g1')).status).toBe(204);
 

tests/postgres/oidcProvider.spec.ts

@@ -51,6 +51,17 @@ describe('oidc_payloads', () => {
     expect(await findOidcPayload('AuthorizationCode', id)).toBeNull();
   });
 
+  it('lets exactly one of many concurrent consumes win', async () => {
+    const id = randomUUID();
+    await upsertOidcPayload('RefreshToken', id, { payload: { grantId: id }, expiresIn: 600 });
+
+    const results = await Promise.all(
+      Array.from({ length: 8 }, () => consumeOidcPayload('RefreshToken', id)),
+    );
+
+    expect(results.filter(Boolean)).toHaveLength(1);
+  });
+
   it('overwrites a record on upsert without losing its consumed state', async () => {
     const id = randomUUID();
 

tests/unit/services/oidcPayloadStore.spec.ts

@@ -267,12 +267,24 @@ describe('findGrantBoundSession', () => {
 });
 
 describe('changing records', () => {
-  it('consumes once', async () => {
-    await consumeOidcPayload('AuthorizationCode', 'c1');
+  it('consumes once, in one conditional statement, and says whether this call did', async () => {
+    vi.mocked(OidcPayload.update).mockResolvedValueOnce([1] as never);
+    expect(await consumeOidcPayload('RefreshToken', 'r1')).toBe(true);
     expect(OidcPayload.update).toHaveBeenCalledWith(
       { consumedAt: expect.any(Date) },
-      { where: { model: 'AuthorizationCode', id: 'c1', consumedAt: { [Op.is]: null } } },
+      {
+        where: {
+          model: 'RefreshToken',
+          id: 'r1',
+          consumedAt: { [Op.is]: null },
+          expiresAt: { [Op.gt]: expect.any(Date) },
+        },
+      },
     );
+
+    // The second of two racing refreshes: the row already had consumedAt.
+    vi.mocked(OidcPayload.update).mockResolvedValueOnce([0] as never);
+    expect(await consumeOidcPayload('RefreshToken', 'r1')).toBe(false);
   });
 
   it('destroys one record, or every record of a grant', async () => {