토너먼트 NPC 개방 베팅을 복구하고 장수 DB 저장을 일괄 처리

2026-09-12 08:04:48 +00:00
parent 3ad230beb3
commit 4dac08b916
9 changed files with 581 additions and 62 deletions
@@ -1,3 +1,4 @@
+import { persistGeneralAccessScores, persistGeneralUpdates } from './generalBatchPersistence.js';
 import { areSeasonRecordsFinalized } from './seasonRecords.js';
 import {
    acquireGameSchemaAdvisoryXactLock,
@@ -1791,32 +1792,26 @@ export const createDatabaseTurnHooks = async (
            }

            await Promise.all([
-                ...generals
-                    .filter((general) => !createdIds.has(general.id))
-                    .map((general) =>
-                        prisma.general.update({
-                            where: { id: general.id },
-                            data: buildGeneralUpdate(general),
-                        })
-                    ),
-                ...generals
-                    .filter(
-                        (general) =>
-                            typeof general.refreshScoreTotal === 'number' && Number.isFinite(general.refreshScoreTotal)
-                    )
-                    .map((general) =>
-                        prisma.generalAccessLog.upsert({
-                            where: { generalId: general.id },
-                            update: {
-                                refreshScoreTotal: Math.floor(general.refreshScoreTotal ?? 0),
-                            },
-                            create: {
-                                generalId: general.id,
-                                userId: general.userId ?? null,
-                                refreshScoreTotal: Math.floor(general.refreshScoreTotal ?? 0),
-                            },
-                        })
-                    ),
+                persistGeneralUpdates(
+                    prisma,
+                    generals
+                        .filter((general) => !createdIds.has(general.id))
+                        .map((general) => ({ id: general.id, data: buildGeneralUpdate(general) }))
+                ),
+                persistGeneralAccessScores(
+                    prisma,
+                    generals
+                        .filter(
+                            (general) =>
+                                typeof general.refreshScoreTotal === 'number' &&
+                                Number.isFinite(general.refreshScoreTotal)
+                        )
+                        .map((general) => ({
+                            generalId: general.id,
+                            userId: general.userId ?? null,
+                            refreshScoreTotal: Math.floor(general.refreshScoreTotal ?? 0),
+                        }))
+                ),
                ...cities.map((city) =>
                    prisma.city.update({
                        where: { id: city.id },
@@ -0,0 +1,100 @@
+import { GamePrisma, type TurnEngineGeneralUpdateInput } from '@sammo-ts/infra';
+
+// Fixed schema identifiers only; all row values remain bound parameters.
+const columns = {
+    userId: 'user_id',
+    name: 'name',
+    nationId: 'nation_id',
+    cityId: 'city_id',
+    troopId: 'troop_id',
+    leadership: 'leadership',
+    strength: 'strength',
+    intel: 'intel',
+    experience: 'experience',
+    dedication: 'dedication',
+    officerLevel: 'officer_level',
+    injury: 'injury',
+    gold: 'gold',
+    rice: 'rice',
+    crew: 'crew',
+    crewTypeId: 'crew_type_id',
+    train: 'train',
+    atmos: 'atmos',
+    age: 'age',
+    npcState: 'npc_state',
+    affinity: 'affinity',
+    bornYear: 'born_year',
+    deadYear: 'dead_year',
+    picture: 'picture',
+    imageServer: 'image_server',
+    startAge: 'start_age',
+    horseCode: 'horse_code',
+    weaponCode: 'weapon_code',
+    bookCode: 'book_code',
+    itemCode: 'item_code',
+    personalCode: 'personal_code',
+    specialCode: 'special_code',
+    special2Code: 'special2_code',
+    lastTurn: 'last_turn',
+    penalty: 'penalty',
+    meta: 'meta',
+    turnTime: 'turn_time',
+    turnTick: 'turn_tick',
+    recentWarTime: 'recent_war_time',
+    recentWarTick: 'recent_war_tick',
+} satisfies Record<keyof TurnEngineGeneralUpdateInput, string>;
+
+export const GENERAL_UPDATE_BATCH_SIZE = 500;
+
+export const persistGeneralUpdates = async (
+    database: { $executeRaw(query: GamePrisma.Sql): Promise<number> },
+    updates: Array<{ id: number; data: TurnEngineGeneralUpdateInput }>
+): Promise<void> => {
+    if (new Set(updates.map((entry) => entry.id)).size !== updates.length) {
+        throw new Error('Duplicate general IDs in persistence batch.');
+    }
+    const assignments = Object.entries(columns).map(([key, column]) => {
+        const identifier = GamePrisma.raw(`"${column}"`);
+        // Prisma omits undefined optional birth/death years instead of clearing them.
+        return key === 'bornYear' || key === 'deadYear'
+            ? GamePrisma.sql`${identifier} = COALESCE(source.${identifier}, target.${identifier})`
+            : GamePrisma.sql`${identifier} = source.${identifier}`;
+    });
+    for (let offset = 0; offset < updates.length; offset += GENERAL_UPDATE_BATCH_SIZE) {
+        const batch = updates.slice(offset, offset + GENERAL_UPDATE_BATCH_SIZE);
+        const rows = batch.map(({ id, data }) => ({
+            id,
+            ...Object.fromEntries(
+                (Object.keys(columns) as Array<keyof typeof columns>).map((key) => [columns[key], data[key]])
+            ),
+        }));
+        const payload = JSON.stringify(rows, (_key, value: unknown) =>
+            typeof value === 'bigint' ? value.toString() : value
+        );
+        const updated = await database.$executeRaw(GamePrisma.sql`
+            UPDATE "general" AS target
+            SET ${GamePrisma.join(assignments)}
+            FROM jsonb_populate_recordset(NULL::"general", ${payload}::jsonb) AS source
+            WHERE target."id" = source."id"
+        `);
+        // Preserve Prisma update's missing-row failure and transaction rollback.
+        if (updated !== batch.length) {
+            throw new Error(`General persistence batch expected ${batch.length} rows, updated ${updated}.`);
+        }
+    }
+};
+
+export const persistGeneralAccessScores = async (
+    database: { $executeRaw(query: GamePrisma.Sql): Promise<number> },
+    updates: Array<{ generalId: number; userId: string | null; refreshScoreTotal: number }>
+): Promise<void> => {
+    for (let offset = 0; offset < updates.length; offset += GENERAL_UPDATE_BATCH_SIZE) {
+        const batch = updates.slice(offset, offset + GENERAL_UPDATE_BATCH_SIZE);
+        await database.$executeRaw(GamePrisma.sql`
+            INSERT INTO "general_access_log" ("general_id", "user_id", "refresh_score_total")
+            VALUES ${GamePrisma.join(batch.map((row) => GamePrisma.sql`(${row.generalId}, ${row.userId}, ${row.refreshScoreTotal})`))}
+            ON CONFLICT ("general_id") DO UPDATE
+            SET "refresh_score_total" = EXCLUDED."refresh_score_total"
+        `);
+    }
+};
@@ -0,0 +1,235 @@
+import { afterAll, beforeAll, describe, expect, it } from 'vitest';
+import { createGamePostgresConnector, type GamePrismaClient, type TurnEngineGeneralUpdateInput } from '@sammo-ts/infra';
+import {
+    GENERAL_UPDATE_BATCH_SIZE,
+    persistGeneralAccessScores,
+    persistGeneralUpdates,
+} from '../src/turn/generalBatchPersistence.js';
+
+const databaseUrl = process.env.INPUT_EVENT_DATABASE_URL;
+const integration = describe.skipIf(!databaseUrl);
+const firstId = 995_000;
+const at = new Date('0200-01-01T00:00:00.000Z');
+
+const dataFor = (index: number): TurnEngineGeneralUpdateInput => ({
+    userId: null,
+    name: `n장 ${index} ' "`,
+    nationId: 0,
+    cityId: 0,
+    troopId: 0,
+    leadership: 51,
+    strength: 52,
+    intel: 53,
+    experience: 123,
+    dedication: 456,
+    officerLevel: 0,
+    injury: 0,
+    gold: 700 + index,
+    rice: 1200 + index,
+    crew: 100,
+    crewTypeId: 1100,
+    train: 70,
+    atmos: 80,
+    age: 21,
+    npcState: 2,
+    affinity: null,
+    bornYear: 179,
+    deadYear: 299,
+    picture: null,
+    imageServer: 0,
+    startAge: 20,
+    horseCode: 'None',
+    weaponCode: 'None',
+    bookCode: 'None',
+    itemCode: 'None',
+    personalCode: 'None',
+    specialCode: 'None',
+    special2Code: 'None',
+    lastTurn: { command: '휴식', args: { text: "한글 ' 문자열" } },
+    penalty: {},
+    meta: { betgold: index * 10, nested: { preserved: true }, list: [1, null, '값'] },
+    turnTime: at,
+    turnTick: 9_007_199_254_740_993n,
+    recentWarTime: index % 2 ? at : null,
+    recentWarTick: index % 2 ? 123n : null,
+});
+
+integration('general batch persistence', () => {
+    let db: GamePrismaClient;
+    let close: () => Promise<void>;
+    beforeAll(async () => {
+        const connector = createGamePostgresConnector({ url: databaseUrl! });
+        await connector.connect();
+        db = connector.prisma;
+        close = () => connector.disconnect();
+        await db.general.createMany({
+            data: Array.from({ length: GENERAL_UPDATE_BATCH_SIZE + 1 }, (_, index) => ({
+                id: firstId + index,
+                name: `original${index}`,
+                turnTime: at,
+                createdAt: at,
+                updatedAt: at,
+            })),
+        });
+    });
+    afterAll(async () => {
+        await db.generalAccessLog.deleteMany({
+            where: { generalId: { gte: firstId, lte: firstId + GENERAL_UPDATE_BATCH_SIZE } },
+        });
+        await db.general.deleteMany({ where: { id: { gte: firstId, lte: firstId + GENERAL_UPDATE_BATCH_SIZE } } });
+        await close();
+    });
+
+    it('matches Prisma row updates including JSON, nulls, dates, bigint and untouched columns', async () => {
+        const data = dataFor(1);
+        await db.general.update({ where: { id: firstId }, data });
+        let writes = 0;
+        await db.$transaction(async (transaction) => {
+            await persistGeneralUpdates(
+                {
+                    $executeRaw: (query) => {
+                        writes += 1;
+                        return transaction.$executeRaw(query);
+                    },
+                },
+                [{ id: firstId + 1, data }]
+            );
+        });
+        const { id: _leftId, ...left } = await db.general.findUniqueOrThrow({ where: { id: firstId } });
+        const { id: _rightId, ...right } = await db.general.findUniqueOrThrow({ where: { id: firstId + 1 } });
+        expect(right).toEqual(left);
+        expect(writes).toBe(1);
+    });
+
+    it('writes 501 distinct generals in two SQL statements and preserves omitted optional years', async () => {
+        const updates = Array.from({ length: GENERAL_UPDATE_BATCH_SIZE + 1 }, (_, index) => ({
+            id: firstId + index,
+            data: { ...dataFor(index), bornYear: undefined, deadYear: undefined },
+        }));
+        let writes = 0;
+        await db.$transaction(async (transaction) => {
+            await persistGeneralUpdates(
+                {
+                    $executeRaw: (query) => {
+                        writes += 1;
+                        return transaction.$executeRaw(query);
+                    },
+                },
+                updates
+            );
+        });
+        expect(writes).toBe(2);
+        const rows = await db.general.findMany({
+            where: { id: { gte: firstId, lte: firstId + GENERAL_UPDATE_BATCH_SIZE } },
+            orderBy: { id: 'asc' },
+        });
+        rows.forEach((row, index) => {
+            expect(row).toMatchObject({
+                gold: 700 + index,
+                rice: 1200 + index,
+                meta: dataFor(index).meta,
+                bornYear: index < 2 ? 179 : 180,
+                deadYear: index < 2 ? 299 : 300,
+                createdAt: at,
+                updatedAt: at,
+                turnTick: 9_007_199_254_740_993n,
+            });
+        });
+    });
+
+    it('rolls back the entire transaction when a target is missing', async () => {
+        const before = await db.general.findUniqueOrThrow({ where: { id: firstId } });
+        await expect(
+            db.$transaction(async (transaction) =>
+                persistGeneralUpdates(transaction, [
+                    { id: firstId, data: dataFor(999) },
+                    { id: firstId - 1, data: dataFor(999) },
+                ])
+            )
+        ).rejects.toThrow('expected 2 rows, updated 1');
+        expect(await db.general.findUniqueOrThrow({ where: { id: firstId } })).toEqual(before);
+    });
+
+    it('preserves unique constraints and rolls back conflicting ownership', async () => {
+        const before = await db.general.findMany({
+            where: { id: { in: [firstId, firstId + 1] } },
+            orderBy: { id: 'asc' },
+        });
+        await expect(
+            db.$transaction(async (transaction) =>
+                persistGeneralUpdates(transaction, [
+                    { id: firstId, data: { ...dataFor(1), userId: 'batch-collision' } },
+                    { id: firstId + 1, data: { ...dataFor(2), userId: 'batch-collision' } },
+                ])
+            )
+        ).rejects.toThrow();
+        expect(
+            await db.general.findMany({ where: { id: { in: [firstId, firstId + 1] } }, orderBy: { id: 'asc' } })
+        ).toEqual(before);
+    });
+
+    it('batch upserts access totals while preserving existing actor and activity fields', async () => {
+        await db.generalAccessLog.create({
+            data: {
+                generalId: firstId,
+                userId: 'original-owner',
+                lastRefresh: at,
+                lastActionAt: at,
+                refresh: 3,
+                refreshTotal: 7,
+                refreshScore: 5,
+            },
+        });
+        const updates = Array.from({ length: GENERAL_UPDATE_BATCH_SIZE + 1 }, (_, index) => ({
+            generalId: firstId + index,
+            userId: null,
+            refreshScoreTotal: index + 10,
+        }));
+        let writes = 0;
+        await db.$transaction(async (transaction) => {
+            await persistGeneralAccessScores(
+                {
+                    $executeRaw: (query) => {
+                        writes += 1;
+                        return transaction.$executeRaw(query);
+                    },
+                },
+                updates
+            );
+        });
+        expect(writes).toBe(2);
+        expect(await db.generalAccessLog.findUniqueOrThrow({ where: { generalId: firstId } })).toMatchObject({
+            userId: 'original-owner',
+            lastRefresh: at,
+            lastActionAt: at,
+            refresh: 3,
+            refreshTotal: 7,
+            refreshScore: 5,
+            refreshScoreTotal: 10,
+        });
+        expect(await db.generalAccessLog.findUniqueOrThrow({ where: { generalId: firstId + 500 } })).toMatchObject({
+            userId: null,
+            lastRefresh: null,
+            refresh: 0,
+            refreshScoreTotal: 510,
+        });
+    });
+
+    it('does not write an empty batch and rejects duplicate IDs before SQL', async () => {
+        let writes = 0;
+        const database = {
+            $executeRaw: async () => {
+                writes += 1;
+                return 0;
+            },
+        };
+        await persistGeneralUpdates(database, []);
+        await expect(
+            persistGeneralUpdates(database, [
+                { id: firstId, data: dataFor(1) },
+                { id: firstId, data: dataFor(2) },
+            ])
+        ).rejects.toThrow('Duplicate general IDs');
+        expect(writes).toBe(0);
+    });
+});