fix reserved turn queue concurrency

2026-07-26 18:32:52 +00:00
parent ab6ed3553a
commit 3cd1ad3b7f
18 changed files with 613 additions and 106 deletions
@@ -0,0 +1,63 @@
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+
+import { createGamePostgresConnector } from '@sammo-ts/infra';
+import type { DatabaseClient } from '../src/context.js';
+import {
+    ReservedTurnRevisionConflictError,
+    getGeneralTurnSnapshot,
+    setGeneralTurn,
+} from '../src/turns/reservedTurns.js';
+
+const databaseUrl = process.env.RESERVED_TURN_DATABASE_URL;
+const describeIntegration = databaseUrl ? describe : describe.skip;
+const GENERAL_ID = 2_147_400_001;
+
+describeIntegration('reserved turn queue revision integration', () => {
+    const connector = databaseUrl ? createGamePostgresConnector({ url: databaseUrl }) : null;
+
+    beforeAll(async () => {
+        if (!connector) {
+            return;
+        }
+        await connector.connect();
+        await connector.prisma.generalTurn.deleteMany({ where: { generalId: GENERAL_ID } });
+        await connector.prisma.generalTurnRevision.deleteMany({ where: { generalId: GENERAL_ID } });
+    });
+
+    afterAll(async () => {
+        if (!connector) {
+            return;
+        }
+        await connector.prisma.generalTurn.deleteMany({ where: { generalId: GENERAL_ID } });
+        await connector.prisma.generalTurnRevision.deleteMany({ where: { generalId: GENERAL_ID } });
+        await connector.disconnect();
+    });
+
+    it('allows exactly one writer for the same expected revision', async () => {
+        if (!connector) {
+            throw new Error('integration connector is unavailable');
+        }
+
+        const write = (action: string) =>
+            connector.prisma.$transaction((transaction) =>
+                setGeneralTurn(transaction as unknown as DatabaseClient, GENERAL_ID, 0, action, {}, 0)
+            );
+
+        const results = await Promise.allSettled([write('che_훈련'), write('che_사기진작')]);
+        const fulfilled = results.filter(
+            (result): result is PromiseFulfilledResult<Awaited<ReturnType<typeof write>>> =>
+                result.status === 'fulfilled'
+        );
+        const rejected = results.filter((result): result is PromiseRejectedResult => result.status === 'rejected');
+
+        expect(fulfilled).toHaveLength(1);
+        expect(fulfilled[0]?.value.revision).toBe(1);
+        expect(rejected).toHaveLength(1);
+        expect(rejected[0]?.reason).toBeInstanceOf(ReservedTurnRevisionConflictError);
+
+        const snapshot = await getGeneralTurnSnapshot(connector.prisma as unknown as DatabaseClient, GENERAL_ID);
+        expect(snapshot.revision).toBe(1);
+        expect(['che_훈련', 'che_사기진작']).toContain(snapshot.turns[0]?.action);
+        expect(await connector.prisma.generalTurn.count({ where: { generalId: GENERAL_ID } })).toBe(30);
+    });
+});
@@ -8,11 +8,14 @@ import {
    setNationTurn,
    shiftGeneralTurns,
    shiftNationTurns,
+    ReservedTurnRevisionConflictError,
 } from '../src/turns/reservedTurns.js';

 const buildDb = () => {
    const generalTurns = new Map<number, GeneralTurnRow[]>();
    const nationTurns = new Map<string, NationTurnRow[]>();
+    const generalRevisions = new Map<number, number>();
+    const nationRevisions = new Map<string, number>();

    type GeneralTurnFindManyArgs = Parameters<DatabaseClient['generalTurn']['findMany']>[0];
    type GeneralTurnDeleteManyArgs = NonNullable<Parameters<DatabaseClient['generalTurn']['deleteMany']>[0]>;
@@ -21,6 +24,16 @@ const buildDb = () => {
    type NationTurnFindManyArgs = Parameters<DatabaseClient['nationTurn']['findMany']>[0];
    type NationTurnDeleteManyArgs = NonNullable<Parameters<DatabaseClient['nationTurn']['deleteMany']>[0]>;
    type NationTurnCreateManyArgs = NonNullable<Parameters<DatabaseClient['nationTurn']['createMany']>[0]>;
+    type GeneralRevisionFindArgs = Parameters<DatabaseClient['generalTurnRevision']['findUnique']>[0];
+    type GeneralRevisionCreateManyArgs = NonNullable<
+        Parameters<DatabaseClient['generalTurnRevision']['createMany']>[0]
+    >;
+    type GeneralRevisionUpdateManyArgs = NonNullable<
+        Parameters<DatabaseClient['generalTurnRevision']['updateMany']>[0]
+    >;
+    type NationRevisionFindArgs = Parameters<DatabaseClient['nationTurnRevision']['findUnique']>[0];
+    type NationRevisionCreateManyArgs = NonNullable<Parameters<DatabaseClient['nationTurnRevision']['createMany']>[0]>;
+    type NationRevisionUpdateManyArgs = NonNullable<Parameters<DatabaseClient['nationTurnRevision']['updateMany']>[0]>;

    const db = {
        worldState: {
@@ -64,6 +77,39 @@ const buildDb = () => {
                return {};
            },
        },
+        generalTurnRevision: {
+            findUnique: async ({ where }: GeneralRevisionFindArgs) => {
+                const generalId = where.generalId as number;
+                const revision = generalRevisions.get(generalId);
+                return revision === undefined
+                    ? null
+                    : {
+                          generalId,
+                          revision,
+                          updatedAt: new Date(),
+                      };
+            },
+            createMany: async ({ data }: GeneralRevisionCreateManyArgs) => {
+                const row = (Array.isArray(data) ? data[0] : data) as {
+                    generalId: number;
+                    revision?: number;
+                };
+                if (generalRevisions.has(row.generalId)) {
+                    return { count: 0 };
+                }
+                generalRevisions.set(row.generalId, row.revision ?? 0);
+                return { count: 1 };
+            },
+            updateMany: async ({ where, data }: GeneralRevisionUpdateManyArgs) => {
+                const generalId = typeof where?.generalId === 'number' ? where.generalId : -1;
+                const expected = typeof where?.revision === 'number' ? where.revision : -1;
+                if (generalRevisions.get(generalId) !== expected || typeof data.revision !== 'number') {
+                    return { count: 0 };
+                }
+                generalRevisions.set(generalId, data.revision);
+                return { count: 1 };
+            },
+        },
        nationTurn: {
            findMany: async (args?: NationTurnFindManyArgs) => {
                const nationId = typeof args?.where?.nationId === 'number' ? args.where.nationId : undefined;
@@ -100,6 +146,48 @@ const buildDb = () => {
                return {};
            },
        },
+        nationTurnRevision: {
+            findUnique: async ({ where }: NationRevisionFindArgs) => {
+                const compound = where.nationId_officerLevel;
+                if (!compound) {
+                    return null;
+                }
+                const key = `${compound.nationId}:${compound.officerLevel}`;
+                const revision = nationRevisions.get(key);
+                return revision === undefined
+                    ? null
+                    : {
+                          nationId: compound.nationId,
+                          officerLevel: compound.officerLevel,
+                          revision,
+                          updatedAt: new Date(),
+                      };
+            },
+            createMany: async ({ data }: NationRevisionCreateManyArgs) => {
+                const row = (Array.isArray(data) ? data[0] : data) as {
+                    nationId: number;
+                    officerLevel: number;
+                    revision?: number;
+                };
+                const key = `${row.nationId}:${row.officerLevel}`;
+                if (nationRevisions.has(key)) {
+                    return { count: 0 };
+                }
+                nationRevisions.set(key, row.revision ?? 0);
+                return { count: 1 };
+            },
+            updateMany: async ({ where, data }: NationRevisionUpdateManyArgs) => {
+                const nationId = typeof where?.nationId === 'number' ? where.nationId : -1;
+                const officerLevel = typeof where?.officerLevel === 'number' ? where.officerLevel : -1;
+                const expected = typeof where?.revision === 'number' ? where.revision : -1;
+                const key = `${nationId}:${officerLevel}`;
+                if (nationRevisions.get(key) !== expected || typeof data.revision !== 'number') {
+                    return { count: 0 };
+                }
+                nationRevisions.set(key, data.revision);
+                return { count: 1 };
+            },
+        },
    } as unknown as DatabaseClient;

    return { db };
@@ -109,30 +197,45 @@ describe('reservedTurns', () => {
    it('sets and shifts general turns', async () => {
        const { db } = buildDb();

-        const initial = await setGeneralTurn(db, 1, 0, 'che_화계', { destCityId: 10 });
+        const initial = await setGeneralTurn(db, 1, 0, 'che_화계', { destCityId: 10 }, 0);

-        expect(initial).toHaveLength(MAX_GENERAL_TURNS);
-        expect(initial[0]?.action).toBe('che_화계');
+        expect(initial.revision).toBe(1);
+        expect(initial.turns).toHaveLength(MAX_GENERAL_TURNS);
+        expect(initial.turns[0]?.action).toBe('che_화계');

-        const pushed = await shiftGeneralTurns(db, 1, 1);
-        expect(pushed[0]?.action).toBe('휴식');
-        expect(pushed[1]?.action).toBe('che_화계');
+        const pushed = await shiftGeneralTurns(db, 1, 1, initial.revision);
+        expect(pushed.revision).toBe(2);
+        expect(pushed.turns[0]?.action).toBe('휴식');
+        expect(pushed.turns[1]?.action).toBe('che_화계');

-        const pulled = await shiftGeneralTurns(db, 1, -1);
-        expect(pulled[0]?.action).toBe('che_화계');
-        expect(pulled[MAX_GENERAL_TURNS - 1]?.action).toBe('휴식');
+        const pulled = await shiftGeneralTurns(db, 1, -1, pushed.revision);
+        expect(pulled.turns[0]?.action).toBe('che_화계');
+        expect(pulled.turns[MAX_GENERAL_TURNS - 1]?.action).toBe('휴식');
+
+        await expect(setGeneralTurn(db, 1, 2, 'che_훈련', {}, 1)).rejects.toBeInstanceOf(
+            ReservedTurnRevisionConflictError
+        );
    });

    it('sets and shifts nation turns', async () => {
        const { db } = buildDb();

-        const initial = await setNationTurn(db, 2, 5, 0, 'che_포상', { isGold: true, amount: 200, destGeneralId: 7 });
+        const initial = await setNationTurn(
+            db,
+            2,
+            5,
+            0,
+            'che_포상',
+            { isGold: true, amount: 200, destGeneralId: 7 },
+            0
+        );

-        expect(initial).toHaveLength(MAX_NATION_TURNS);
-        expect(initial[0]?.action).toBe('che_포상');
+        expect(initial.revision).toBe(1);
+        expect(initial.turns).toHaveLength(MAX_NATION_TURNS);
+        expect(initial.turns[0]?.action).toBe('che_포상');

-        const pushed = await shiftNationTurns(db, 2, 5, 1);
-        expect(pushed[0]?.action).toBe('휴식');
-        expect(pushed[1]?.action).toBe('che_포상');
+        const pushed = await shiftNationTurns(db, 2, 5, 1, initial.revision);
+        expect(pushed.turns[0]?.action).toBe('휴식');
+        expect(pushed.turns[1]?.action).toBe('che_포상');
    });
 });
@@ -101,6 +101,8 @@ const buildContext = (options?: {
    const battleSim = options?.battleSim ?? new InMemoryBattleSimTransport();
    const generalTurns = options?.generalTurns ?? [];
    const nationTurns = options?.nationTurns ?? [];
+    let generalTurnRevision: number | undefined;
+    let nationTurnRevision: number | undefined;
    const db = {
        worldState: {
            findFirst: async () => {
@@ -133,17 +135,60 @@ const buildContext = (options?: {
                return {};
            },
        },
+        generalTurnRevision: {
+            findUnique: async () =>
+                generalTurnRevision === undefined
+                    ? null
+                    : { generalId: options?.general?.id ?? 0, revision: generalTurnRevision, updatedAt: new Date() },
+            createMany: async ({ data }: { data: Array<{ revision: number }> }) => {
+                if (generalTurnRevision !== undefined) {
+                    return { count: 0 };
+                }
+                generalTurnRevision = data[0]?.revision ?? 0;
+                return { count: 1 };
+            },
+            updateMany: async ({ where, data }: { where: { revision: number }; data: { revision: number } }) => {
+                if (generalTurnRevision !== where.revision) {
+                    return { count: 0 };
+                }
+                generalTurnRevision = data.revision;
+                return { count: 1 };
+            },
+        },
        nationTurn: {
            findMany: async ({ where }: { where: { nationId: number; officerLevel: number } }) =>
-                nationTurns.filter(
-                    (row) => row.nationId === where.nationId && row.officerLevel === where.officerLevel
-                ),
+                nationTurns.filter((row) => row.nationId === where.nationId && row.officerLevel === where.officerLevel),
            deleteMany: async () => ({}),
            createMany: async (args: unknown) => {
                options?.nationTurnWrites?.push(args);
                return {};
            },
        },
+        nationTurnRevision: {
+            findUnique: async () =>
+                nationTurnRevision === undefined
+                    ? null
+                    : {
+                          nationId: options?.general?.nationId ?? 0,
+                          officerLevel: options?.general?.officerLevel ?? 0,
+                          revision: nationTurnRevision,
+                          updatedAt: new Date(),
+                      },
+            createMany: async ({ data }: { data: Array<{ revision: number }> }) => {
+                if (nationTurnRevision !== undefined) {
+                    return { count: 0 };
+                }
+                nationTurnRevision = data[0]?.revision ?? 0;
+                return { count: 1 };
+            },
+            updateMany: async ({ where, data }: { where: { revision: number }; data: { revision: number } }) => {
+                if (nationTurnRevision !== where.revision) {
+                    return { count: 0 };
+                }
+                nationTurnRevision = data.revision;
+                return { count: 1 };
+            },
+        },
    };
    const accessTokenStore = new RedisAccessTokenStore(
        {
@@ -370,8 +415,9 @@ describe('appRouter', () => {
        const caller = appRouter.createCaller(buildContext({ general, generalTurns }));
        const response = await caller.turns.reserved.getGeneral({ generalId: 11 });

-        expect(response[0]?.action).toBe('che_화계');
-        expect(response[0]?.index).toBe(0);
+        expect(response.revision).toBe(0);
+        expect(response.turns[0]?.action).toBe('che_화계');
+        expect(response.turns[0]?.index).toBe(0);
    });

    it('returns reserved nation turns', async () => {
@@ -390,8 +436,9 @@ describe('appRouter', () => {
        const caller = appRouter.createCaller(buildContext({ general, nationTurns }));
        const response = await caller.turns.reserved.getNation({ generalId: 12 });

-        expect(response[0]?.action).toBe('che_포상');
-        expect(response[0]?.index).toBe(0);
+        expect(response.revision).toBe(0);
+        expect(response.turns[0]?.action).toBe('che_포상');
+        expect(response.turns[0]?.index).toBe(0);
    });

    it('validates and persists general command arguments from the authenticated owner', async () => {
@@ -404,6 +451,7 @@ describe('appRouter', () => {
            turnIndex: 0,
            action: 'che_화계',
            args: { destCityId: 7 },
+            expectedRevision: 0,
        });

        expect(response.turns[0]).toMatchObject({ action: 'che_화계', args: { destCityId: 7 } });
@@ -416,6 +464,16 @@ describe('appRouter', () => {
            actionCode: 'che_화계',
            arg: { destCityId: 7 },
        });
+
+        await expect(
+            caller.turns.reserved.setGeneral({
+                generalId: 13,
+                turnIndex: 1,
+                action: '휴식',
+                expectedRevision: 0,
+            })
+        ).rejects.toMatchObject({ code: 'CONFLICT' });
+        expect(writes).toHaveLength(1);
    });

    it('rejects malformed and cross-scope arguments without writing turns', async () => {
@@ -432,6 +490,7 @@ describe('appRouter', () => {
                turnIndex: 0,
                action: 'che_화계',
                args: { destCityId: '7' },
+                expectedRevision: 0,
            })
        ).rejects.toMatchObject({ code: 'BAD_REQUEST' });
        await expect(
@@ -440,6 +499,7 @@ describe('appRouter', () => {
                turnIndex: 0,
                action: 'che_포상',
                args: { isGold: true, amount: 1, destGeneralId: 7 },
+                expectedRevision: 0,
            })
        ).rejects.toMatchObject({ code: 'BAD_REQUEST' });

@@ -465,6 +525,7 @@ describe('appRouter', () => {
                generalId: general.id,
                turnIndex: 0,
                action: '휴식',
+                expectedRevision: 0,
            })
        ).rejects.toMatchObject({
            code: 'FORBIDDEN',
@@ -473,6 +534,7 @@ describe('appRouter', () => {
            caller.turns.reserved.shiftGeneral({
                generalId: general.id,
                amount: 1,
+                expectedRevision: 0,
            })
        ).rejects.toMatchObject({
            code: 'FORBIDDEN',
@@ -482,6 +544,7 @@ describe('appRouter', () => {
                generalId: general.id,
                turnIndex: 0,
                action: '휴식',
+                expectedRevision: 0,
            })
        ).rejects.toMatchObject({
            code: 'FORBIDDEN',
@@ -490,6 +553,7 @@ describe('appRouter', () => {
            caller.turns.reserved.shiftNation({
                generalId: general.id,
                amount: 1,
+                expectedRevision: 0,
            })
        ).rejects.toMatchObject({
            code: 'FORBIDDEN',