diff --git a/app/game-api/src/maps/worldMap.ts b/app/game-api/src/maps/worldMap.ts index 94c297b6..18289014 100644 --- a/app/game-api/src/maps/worldMap.ts +++ b/app/game-api/src/maps/worldMap.ts @@ -1,5 +1,5 @@ import type { GameApiContext, WorldStateRow } from '../context.js'; -import { asRecord, isRecord } from '@sammo-ts/common'; +import { asRecord, buildGameReadModelDomainRevisionKey, isRecord } from '@sammo-ts/common'; export type MapCityCompact = [number, number, number, number, number, number]; export type MapNationCompact = [number, string, string, number]; @@ -94,6 +94,24 @@ const resolveSpyList = (meta: Record): Record = const buildBaseMapCacheKey = (ctx: GameApiContext, scope: 'base' | 'public' = 'base'): string => `sammo:map:${scope}:${ctx.profile.id}:${ctx.profile.scenario}`; +const loadWorldMapRevision = async (ctx: GameApiContext): Promise => { + const redis = ctx.redis as unknown as { + hGet?: (key: string, field: string) => Promise; + }; + if (typeof redis.hGet !== 'function') { + return '0'; + } + try { + return (await redis.hGet(buildGameReadModelDomainRevisionKey(ctx.profile.name), 'world')) ?? '0'; + } catch { + // Cache revision lookup must not make the map unavailable. + return '0'; + } +}; + +export const buildRevisionedBaseMapCacheKey = async (ctx: GameApiContext): Promise => + `${buildBaseMapCacheKey(ctx)}:r${await loadWorldMapRevision(ctx)}`; + const loadBaseMap = async ( ctx: GameApiContext, options?: { @@ -103,7 +121,7 @@ const loadBaseMap = async ( } ): Promise => { const useCache = options?.useCache ?? true; - const cacheKey = options?.cacheKey ?? buildBaseMapCacheKey(ctx); + const cacheKey = options?.cacheKey ?? (await buildRevisionedBaseMapCacheKey(ctx)); const ttlSeconds = options?.ttlSeconds ?? BASE_MAP_TTL_SECONDS; if (useCache) { diff --git a/app/game-api/test/worldMapRevisionCache.test.ts b/app/game-api/test/worldMapRevisionCache.test.ts new file mode 100644 index 00000000..1c812edc --- /dev/null +++ b/app/game-api/test/worldMapRevisionCache.test.ts @@ -0,0 +1,36 @@ +import { describe, expect, it } from 'vitest'; +import { buildGameReadModelDomainRevisionKey } from '@sammo-ts/common'; + +import { buildRevisionedBaseMapCacheKey } from '../src/maps/worldMap.js'; +import type { GameApiContext } from '../src/context.js'; + +describe('world map revision cache', () => { + it('selects a new shared base-map key after a committed world revision', async () => { + const reads: Array<[string, string]> = []; + const ctx = { + profile: { id: 'hwe', name: 'hwe', scenario: 'scenario_2400' }, + redis: { + hGet: async (key: string, field: string) => { + reads.push([key, field]); + return '12'; + }, + }, + } as unknown as GameApiContext; + + await expect(buildRevisionedBaseMapCacheKey(ctx)).resolves.toBe( + 'sammo:map:base:hwe:scenario_2400:r12' + ); + expect(reads).toEqual([[buildGameReadModelDomainRevisionKey('hwe'), 'world']]); + }); + + it('falls back to revision zero when Redis is temporarily unavailable', async () => { + const ctx = { + profile: { id: 'hwe', name: 'hwe', scenario: 'scenario_2400' }, + redis: { hGet: async () => Promise.reject(new Error('redis unavailable')) }, + } as unknown as GameApiContext; + + await expect(buildRevisionedBaseMapCacheKey(ctx)).resolves.toBe( + 'sammo:map:base:hwe:scenario_2400:r0' + ); + }); +}); diff --git a/app/game-engine/src/lifecycle/turnDaemonLifecycle.ts b/app/game-engine/src/lifecycle/turnDaemonLifecycle.ts index f56702f5..a7e4d5a5 100644 --- a/app/game-engine/src/lifecycle/turnDaemonLifecycle.ts +++ b/app/game-engine/src/lifecycle/turnDaemonLifecycle.ts @@ -351,6 +351,14 @@ export class TurnDaemonLifecycle { await this.resolveNextRunTime(); } + try { + await this.hooks?.publishCommandEvents?.(result); + } catch (error) { + // The command is already durable. Realtime publication is a + // best-effort read-model invalidation and must not reject it. + this.status.lastError = error instanceof Error ? error.message : 'Unknown command event publication error.'; + } + if (this.commandResponder && command.requestId) { await this.commandResponder.publishCommandResult(command.requestId, result); } diff --git a/app/game-engine/src/lifecycle/types.ts b/app/game-engine/src/lifecycle/types.ts index 8389d788..507c8004 100644 --- a/app/game-engine/src/lifecycle/types.ts +++ b/app/game-engine/src/lifecycle/types.ts @@ -70,6 +70,7 @@ export interface TurnDaemonHooks { requestId: string, execute: (context: TurnDaemonCommandExecutionContext) => Promise ): Promise; + publishCommandEvents?(result: TurnDaemonCommandResult): Promise; publishEvents?(result: TurnRunResult): Promise; onRunError?(error: unknown): Promise; } diff --git a/app/game-engine/src/turn/databaseHooks.ts b/app/game-engine/src/turn/databaseHooks.ts index b55e94d8..fd7fc736 100644 --- a/app/game-engine/src/turn/databaseHooks.ts +++ b/app/game-engine/src/turn/databaseHooks.ts @@ -22,10 +22,10 @@ import { type LogEntryDraft, type MessageRecordDraft, } from '@sammo-ts/logic'; -import { asRecord } from '@sammo-ts/common'; +import { asRecord, type RealtimeReadModelChanges } from '@sammo-ts/common'; import type { TurnDaemonCommandResult, TurnDaemonHooks } from '../lifecycle/types.js'; -import type { InMemoryTurnWorld } from './inMemoryWorld.js'; +import type { InMemoryTurnWorld, TurnWorldChanges } from './inMemoryWorld.js'; import type { InMemoryReservedTurnStore, ReservedTurnChanges } from './reservedTurnStore.js'; import { buildDiplomacyMeta } from '@sammo-ts/logic'; import { ensureItemInventory, withSerializedItemInventory } from '@sammo-ts/logic/items/index.js'; @@ -40,9 +40,85 @@ import { persistYearbookSnapshot } from './yearbookPersistence.js'; export interface DatabaseTurnHooks { hooks: TurnDaemonHooks; + takeCommittedReadModelChanges(): RealtimeReadModelChanges | null; close(): Promise; } +const uniqueSortedIds = (values: Iterable): number[] => + [...new Set(values)].filter((value) => Number.isSafeInteger(value) && value > 0).sort((left, right) => left - right); + +export const summarizeRealtimeReadModelChanges = ( + changes: TurnWorldChanges, + reservedTurnChanges?: ReservedTurnChanges +): RealtimeReadModelChanges => { + const generalIds = uniqueSortedIds([ + ...changes.generals.map((general) => general.id), + ...changes.createdGenerals.map((general) => general.id), + ...changes.deletedGenerals, + ...changes.lifecycleEvents.map((event) => event.generalId), + ]); + const cityIds = uniqueSortedIds(changes.cities.map((city) => city.id)); + const nationIds = uniqueSortedIds([ + ...changes.nations.map((nation) => nation.id), + ...changes.createdNations.map((nation) => nation.id), + ...changes.deletedNations, + ...changes.deletedNationSnapshots.map((snapshot) => snapshot.nation.id), + ]); + const reservedGeneralIds = uniqueSortedIds( + reservedTurnChanges + ? [ + ...reservedTurnChanges.generalIds, + ...reservedTurnChanges.generalInitializationIds, + ...reservedTurnChanges.generalLeaseIds, + ] + : [] + ); + const recordGeneralIds = uniqueSortedIds( + changes.logs.flatMap((entry) => + entry.scope === LogScope.GENERAL && entry.category === LogCategory.ACTION && entry.generalId + ? [entry.generalId] + : [] + ) + ); + const globalRecordsChanged = changes.logs.some( + (entry) => + entry.scope === LogScope.SYSTEM && + (entry.category === LogCategory.SUMMARY || entry.category === LogCategory.ACTION) + ); + const worldHistoryChanged = changes.logs.some( + (entry) => entry.scope === LogScope.SYSTEM && entry.category === LogCategory.HISTORY + ); + const contactsChanged = + changes.createdGenerals.length > 0 || + changes.deletedGenerals.length > 0 || + changes.createdNations.length > 0 || + changes.deletedNations.length > 0 || + changes.lifecycleEvents.some((event) => { + const after = event.after; + const beforePermission = asRecord(event.before.meta).permission; + const afterPermission = after ? asRecord(after.meta).permission : undefined; + return ( + !after || + event.before.name !== after.name || + event.before.nationId !== after.nationId || + event.before.officerLevel !== after.officerLevel || + beforePermission !== afterPermission + ); + }); + + return { + generalIds, + cityIds, + nationIds, + reservedGeneralIds, + recordGeneralIds, + worldChanged: false, + globalRecordsChanged, + worldHistoryChanged, + contactsChanged, + }; +}; + export const excludeDeletedReservedTurnQueues = ( changes: ReservedTurnChanges, deletedGeneralIds: readonly number[], @@ -588,11 +664,12 @@ export const createDatabaseTurnHooks = async ( const connector = createGamePostgresConnector({ url: databaseUrl }); await connector.connect(); const prisma = connector.prisma; + let committedReadModelChanges: RealtimeReadModelChanges | null = null; const persistChanges = async ( transaction?: GamePrisma.TransactionClient, commandCompletion?: { requestId: string; result: TurnDaemonCommandResult } - ): Promise<() => void> => { + ): Promise<{ acknowledge: () => void; readModelChanges: RealtimeReadModelChanges }> => { const state = world.getState(); const changes = world.peekDirtyState(); const { @@ -1059,39 +1136,50 @@ export const createDatabaseTurnHooks = async ( ); } - return () => { - world.acknowledgeDirtyState(changes); - if (options?.reservedTurns && reservedTurnChanges) { - options.reservedTurns.acknowledgeDirtyState(reservedTurnChanges); - } + return { + acknowledge: () => { + world.acknowledgeDirtyState(changes); + if (options?.reservedTurns && reservedTurnChanges) { + options.reservedTurns.acknowledgeDirtyState(reservedTurnChanges); + } + }, + readModelChanges: summarizeRealtimeReadModelChanges(changes, persistedReservedTurnChanges), }; }; const hooks: TurnDaemonHooks = { flushChanges: async () => { - const acknowledge = await persistChanges(); - acknowledge(); + const committed = await persistChanges(); + committed.acknowledge(); + committedReadModelChanges = committed.readModelChanges; }, commitCommand: async (requestId, result) => { - const acknowledge = await persistChanges(undefined, { requestId, result }); - acknowledge(); + const committed = await persistChanges(undefined, { requestId, result }); + committed.acknowledge(); + committedReadModelChanges = committed.readModelChanges; }, executeCommand: async (requestId, execute) => { const committed = await prisma.$transaction( async (transaction) => { const result = await execute({ db: transaction }); - const acknowledge = await persistChanges(transaction, { requestId, result }); - return { result, acknowledge }; + const persisted = await persistChanges(transaction, { requestId, result }); + return { result, persisted }; }, options?.transactionTimeoutMs ? { timeout: options.transactionTimeoutMs } : undefined ); - committed.acknowledge(); + committed.persisted.acknowledge(); + committedReadModelChanges = committed.persisted.readModelChanges; return committed.result; }, }; return { hooks, + takeCommittedReadModelChanges: () => { + const changes = committedReadModelChanges; + committedReadModelChanges = null; + return changes; + }, close: () => connector.disconnect(), }; }; diff --git a/app/game-engine/src/turn/turnDaemon.ts b/app/game-engine/src/turn/turnDaemon.ts index d594cdd1..666376f9 100644 --- a/app/game-engine/src/turn/turnDaemon.ts +++ b/app/game-engine/src/turn/turnDaemon.ts @@ -1,5 +1,15 @@ import { loadActionModuleBundle, type TurnCommandProfile, type TurnSchedule } from '@sammo-ts/logic'; -import { buildGameEventChannel, GameClock, type GameClockMode, type RealtimeEvent } from '@sammo-ts/common'; +import { + buildGameEventChannel, + buildGameReadModelDomainRevisionKey, + buildGameReadModelRevisionKey, + createEmptyRealtimeReadModelChanges, + GameClock, + hasRealtimeReadModelChanges, + type GameClockMode, + type RealtimeEvent, + type RealtimeReadModelChanges, +} from '@sammo-ts/common'; import { createGamePostgresConnector, createRedisConnector, resolveRedisConfigFromEnv } from '@sammo-ts/infra'; import { NATION_TRAIT_KEYS, NationTraitLoader, loadNationTraitModules } from '@sammo-ts/logic'; @@ -633,6 +643,8 @@ const createTurnDaemonRuntimeWithLease = async ( let hooks: TurnDaemonHooks | undefined; let publishRealtimeEvent: ((event: RealtimeEvent) => Promise) | null = null; + let publishReadModelChanges: ((changes: RealtimeReadModelChanges) => Promise) | null = null; + let takeCommittedReadModelChanges: (() => RealtimeReadModelChanges | null) | null = null; let close = async () => {}; let auctionFinalizer: Awaited> | null = null; let auctionBidder: Awaited> | null = null; @@ -676,6 +688,7 @@ const createTurnDaemonRuntimeWithLease = async ( await gatewayGate?.markPaused(error); }, }; + takeCommittedReadModelChanges = dbHooks.takeCommittedReadModelChanges; close = async () => { if (auctionBidder) { await auctionBidder.close(); @@ -730,25 +743,66 @@ const createTurnDaemonRuntimeWithLease = async ( publishRealtimeEvent = async (event: RealtimeEvent) => { await redisClient.publish(realtimeChannel, JSON.stringify(event)); }; + const revisionKey = buildGameReadModelRevisionKey(options.profileName ?? options.profile); + const domainRevisionKey = buildGameReadModelDomainRevisionKey(options.profileName ?? options.profile); + publishReadModelChanges = async (changes) => { + if (changes.worldChanged || changes.cityIds.length > 0 || changes.nationIds.length > 0) { + await redisClient.hIncrBy(domainRevisionKey, 'world', 1); + } + return redisClient.incr(revisionKey); + }; } if (publishRealtimeEvent) { const basePublishEvents = hooks?.publishEvents; - // 턴 처리 완료 이벤트를 실시간 채널로 전파한다. + const basePublishCommandEvents = hooks?.publishCommandEvents; + const publishCommittedChanges = async (changes: RealtimeReadModelChanges): Promise => { + if (!hasRealtimeReadModelChanges(changes)) { + return undefined; + } + return publishReadModelChanges?.(changes); + }; + // Durable mutation summaries invalidate only the affected read models. hooks = { ...hooks, publishEvents: async (result) => { try { + const changes = takeCommittedReadModelChanges?.() ?? createEmptyRealtimeReadModelChanges(); + if (result.processedTurns > 0) { + changes.worldChanged = true; + } + const revision = await publishCommittedChanges(changes); await publishRealtimeEvent({ type: 'turnCompleted', at: new Date().toISOString(), lastTurnTime: result.lastTurnTime, + changes, + revision, }); } catch { // 실시간 이벤트 전송 실패는 턴 처리 결과에 영향을 주지 않는다. } await basePublishEvents?.(result); }, + publishCommandEvents: async (result) => { + try { + const changes = takeCommittedReadModelChanges?.(); + if (changes && hasRealtimeReadModelChanges(changes)) { + const revision = await publishCommittedChanges(changes); + if (revision !== undefined) { + await publishRealtimeEvent({ + type: 'readModelChanged', + at: new Date().toISOString(), + changes, + revision, + }); + } + } + } catch { + // 명령은 이미 commit되었으므로 이벤트 실패로 되돌리지 않는다. + } + await basePublishCommandEvents?.(result); + }, }; } diff --git a/app/game-engine/test/inputEventAtomicity.test.ts b/app/game-engine/test/inputEventAtomicity.test.ts index 99f51f49..f28ab2e0 100644 --- a/app/game-engine/test/inputEventAtomicity.test.ts +++ b/app/game-engine/test/inputEventAtomicity.test.ts @@ -156,6 +156,9 @@ describe('input event atomicity', () => { commitCommand: async (requestId, committedResult) => { order.push(`commit:${requestId}:${committedResult.type}`); }, + publishCommandEvents: async (committedResult) => { + order.push(`publish:${committedResult.type}`); + }, }, commandResponder: { publishStatus: async () => {}, @@ -181,7 +184,12 @@ describe('input event atomicity', () => { const loop = lifecycle.start(); await responded; - expect(order).toEqual(['handle:auctionBid', 'commit:event-1:auctionBid', 'respond:event-1:auctionBid']); + expect(order).toEqual([ + 'handle:auctionBid', + 'commit:event-1:auctionBid', + 'publish:auctionBid', + 'respond:event-1:auctionBid', + ]); await lifecycle.stop('done'); await loop; diff --git a/app/game-engine/test/realtimeReadModelChanges.test.ts b/app/game-engine/test/realtimeReadModelChanges.test.ts new file mode 100644 index 00000000..426d025d --- /dev/null +++ b/app/game-engine/test/realtimeReadModelChanges.test.ts @@ -0,0 +1,63 @@ +import { describe, expect, it } from 'vitest'; +import { LogCategory, LogFormat, LogScope } from '@sammo-ts/logic'; + +import { summarizeRealtimeReadModelChanges } from '../src/turn/databaseHooks.js'; +import type { TurnWorldChanges } from '../src/turn/inMemoryWorld.js'; +import type { ReservedTurnChanges } from '../src/turn/reservedTurnStore.js'; + +describe('summarizeRealtimeReadModelChanges', () => { + it('emits deterministic entity and record invalidations from committed changes', () => { + const worldChanges = { + generals: [{ id: 9 }, { id: 7 }], + createdGenerals: [{ id: 8 }], + deletedGenerals: [9], + cities: [{ id: 4 }], + nations: [{ id: 3 }], + createdNations: [], + deletedNations: [], + deletedNationSnapshots: [], + lifecycleEvents: [], + logs: [ + { + scope: LogScope.GENERAL, + category: LogCategory.ACTION, + generalId: 7, + format: LogFormat.PLAIN, + text: 'general', + }, + { + scope: LogScope.SYSTEM, + category: LogCategory.SUMMARY, + format: LogFormat.PLAIN, + text: 'summary', + }, + { + scope: LogScope.SYSTEM, + category: LogCategory.HISTORY, + format: LogFormat.PLAIN, + text: 'history', + }, + ], + } as unknown as TurnWorldChanges; + const reservedChanges: ReservedTurnChanges = { + generalIds: [9], + generalInitializationIds: [7], + generalLeaseIds: [9, 8], + nationKeys: [], + nationInitializationKeys: [], + nationLeaseKeys: [], + }; + + expect(summarizeRealtimeReadModelChanges(worldChanges, reservedChanges)).toEqual({ + generalIds: [7, 8, 9], + cityIds: [4], + nationIds: [3], + reservedGeneralIds: [7, 8, 9], + recordGeneralIds: [7], + worldChanged: false, + globalRecordsChanged: true, + worldHistoryChanged: true, + contactsChanged: true, + }); + }); +}); diff --git a/app/game-frontend/e2e/mainNavigation.spec.ts b/app/game-frontend/e2e/mainNavigation.spec.ts index 0a938592..56beec4d 100644 --- a/app/game-frontend/e2e/mainNavigation.spec.ts +++ b/app/game-frontend/e2e/mainNavigation.spec.ts @@ -517,7 +517,7 @@ test('mobile single document refreshes once and preserves tokens on lobby return expect(state.operations).not.toContain('auth.logout'); }); -test('turn realtime refresh is rate limited, patches in place, and stops after leaving main', async ({ page }) => { +test('realtime read-model events skip clock-only work, merge bursts, patch in place, and stop off-route', async ({ page }) => { const state: NavigationFixture = { officerLevel: 5, permission: 2, @@ -564,23 +564,74 @@ test('turn realtime refresh is rate limited, patches in place, and stops after l }); const callsBeforeRefresh = state.generalMeCalls; + const operationsBeforeClockOnly = state.operations.length; + await page.evaluate(() => { + (window as unknown as { __emitMainRealtime: (type: string, payload: unknown) => void }).__emitMainRealtime( + 'turnCompleted', + { + at: new Date().toISOString(), + lastTurnTime: '0185-02-01T00:00:00.000Z', + changes: { + generalIds: [], + cityIds: [], + nationIds: [], + reservedGeneralIds: [], + recordGeneralIds: [], + worldChanged: false, + globalRecordsChanged: false, + worldHistoryChanged: false, + contactsChanged: false, + }, + } + ); + }); + await new Promise((resolve) => setTimeout(resolve, 300)); + expect(state.operations.slice(operationsBeforeClockOnly)).toEqual([]); + + const operationsBeforeChangedBurst = state.operations.length; state.generalName = '부드럽게갱신된장수'; await page.evaluate(() => { const emit = (window as unknown as { __emitMainRealtime: (type: string, payload: unknown) => void }) .__emitMainRealtime; for (let index = 0; index < 100; index += 1) { - emit('turnCompleted', { at: new Date().toISOString(), lastTurnTime: '0185-02-01T00:00:00.000Z' }); + emit('turnCompleted', { + at: new Date().toISOString(), + lastTurnTime: '0185-02-01T00:00:00.000Z', + changes: { + generalIds: [7], + cityIds: [], + nationIds: [], + reservedGeneralIds: [], + recordGeneralIds: [], + worldChanged: false, + globalRecordsChanged: false, + worldHistoryChanged: false, + contactsChanged: false, + }, + }); } }); - await new Promise((resolve) => setTimeout(resolve, 500)); - expect(state.generalMeCalls).toBe(callsBeforeRefresh); - await expect.poll(() => state.generalMeCalls, { timeout: 7_000 }).toBe(callsBeforeRefresh + 1); + await expect.poll(() => state.generalMeCalls, { timeout: 3_000 }).toBe(callsBeforeRefresh + 1); await expect(page.locator('[data-main-target="general"] .skeleton-line')).toHaveCount(0); await expect(page.locator('[data-main-target="city"] .skeleton-line')).toHaveCount(0); await expect(page.getByRole('button', { name: '갱 신' })).toHaveAttribute('aria-busy', 'false'); expect(state.generalMeCalls).toBe(callsBeforeRefresh + 1); await expect(page.locator('.general-title')).toContainText('부드럽게갱신된장수'); + const changedOperations = state.operations.slice(operationsBeforeChangedBurst); + expect(changedOperations).toEqual( + expect.arrayContaining(['general.me', 'world.getMap', 'turns.getCommandTable', 'board.getAccess']) + ); + expect(changedOperations).not.toEqual( + expect.arrayContaining([ + 'lobby.info', + 'messages.getRecent', + 'messages.getContacts', + 'general.getRecentRecords', + 'general.getFrontStatus', + 'turns.reserved.getGeneral', + ]) + ); const profile = await page.evaluate(() => { const probe = ( @@ -616,7 +667,7 @@ test('turn realtime refresh is rate limited, patches in place, and stops after l `${JSON.stringify( { emittedTurnEvents: 100, - refreshRequests: state.generalMeCalls - callsBeforeRefresh, + selectiveGeneralRefreshes: state.generalMeCalls - callsBeforeRefresh, inFlightSkeletons: { general: 0, city: 0 }, ...profile, }, @@ -639,7 +690,21 @@ test('turn realtime refresh is rate limited, patches in place, and stops after l await page.evaluate(() => { (window as unknown as { __emitMainRealtime: (type: string, payload: unknown) => void }).__emitMainRealtime( 'turnCompleted', - { at: new Date().toISOString(), lastTurnTime: '0185-02-01T00:00:00.000Z' } + { + at: new Date().toISOString(), + lastTurnTime: '0185-02-01T00:00:00.000Z', + changes: { + generalIds: [7], + cityIds: [], + nationIds: [], + reservedGeneralIds: [], + recordGeneralIds: [], + worldChanged: false, + globalRecordsChanged: false, + worldHistoryChanged: false, + contactsChanged: false, + }, + } ); }); await new Promise((resolve) => setTimeout(resolve, 300)); diff --git a/app/game-frontend/src/stores/mainDashboard.ts b/app/game-frontend/src/stores/mainDashboard.ts index 763d564a..aeccfbcd 100644 --- a/app/game-frontend/src/stores/mainDashboard.ts +++ b/app/game-frontend/src/stores/mainDashboard.ts @@ -1,13 +1,14 @@ import { computed, ref, watch } from 'vue'; import { defineStore } from 'pinia'; import { MESSAGE_MAILBOX_NATIONAL_BASE, MESSAGE_MAILBOX_PUBLIC, type MessageType } from '@sammo-ts/logic'; -import type { RealtimeEvent } from '@sammo-ts/common'; +import type { RealtimeEvent, RealtimeReadModelChanges } from '@sammo-ts/common'; import { trpc } from '../utils/trpc'; import { useMapViewerStore } from './mapViewer'; import { useSessionStore } from './session'; import { createLatestRefreshQueue } from '../utils/latestRefreshQueue'; import { createRateLimitedRefreshQueue } from '../utils/rateLimitedRefreshQueue'; import { structurallyShare } from '../utils/structuralShare'; +import { createMergedReadModelRefreshQueue, resolveDashboardRefreshPlan } from '../utils/dashboardReadModel'; const REALTIME_FULL_REFRESH_MIN_INTERVAL_MS = 5_000; @@ -252,6 +253,29 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => { surveyNotice.value = null; }; + const applyRecentRecords = ( + records: Awaited> + ) => { + globalRecords.value = structurallyShare( + globalRecords.value, + mergeRecentRecords(globalRecords.value, records.global) + ); + generalRecords.value = structurallyShare( + generalRecords.value, + mergeRecentRecords(generalRecords.value, records.general) + ); + worldHistory.value = structurallyShare( + worldHistory.value, + mergeRecentRecords(worldHistory.value, records.history) + ); + lastGeneralRecordId = Math.max( + lastGeneralRecordId, + records.global[0]?.id ?? 0, + records.general[0]?.id ?? 0 + ); + lastWorldHistoryId = Math.max(lastWorldHistoryId, records.history[0]?.id ?? 0); + }; + const refreshMainData = async () => { const isInitialLoad = !initialized; if (isInitialLoad) { @@ -337,24 +361,7 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => { ) as ReservedTurnView[]; reservedGeneralRevision.value = generalTurns.revision; if (records) { - globalRecords.value = structurallyShare( - globalRecords.value, - mergeRecentRecords(globalRecords.value, records.global) - ); - generalRecords.value = structurallyShare( - generalRecords.value, - mergeRecentRecords(generalRecords.value, records.general) - ); - worldHistory.value = structurallyShare( - worldHistory.value, - mergeRecentRecords(worldHistory.value, records.history) - ); - lastGeneralRecordId = Math.max( - lastGeneralRecordId, - records.global[0]?.id ?? 0, - records.general[0]?.id ?? 0 - ); - lastWorldHistoryId = Math.max(lastWorldHistoryId, records.history[0]?.id ?? 0); + applyRecentRecords(records); } if (nextFrontStatus) { updateFrontStatus(nextFrontStatus); @@ -381,6 +388,108 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => { minIntervalMs: REALTIME_FULL_REFRESH_MIN_INTERVAL_MS, }); + const refreshChangedReadModels = async (changes: RealtimeReadModelChanges) => { + const id = generalId.value; + if (!id) { + return; + } + const plan = resolveDashboardRefreshPlan(changes, { + generalId: id, + cityId: city.value?.id ?? null, + nationId: nation.value?.id ?? null, + }); + if (!Object.values(plan).some(Boolean)) { + return; + } + + refreshing.value = true; + error.value = null; + if (plan.records) recordsError.value = null; + if (plan.frontStatus) frontStatusError.value = null; + try { + const contextPromise = plan.context + ? trpc.general.me.query() + : Promise.resolve(undefined as GeneralContext | undefined); + const lobbyPromise = plan.lobby ? trpc.lobby.info.query() : Promise.resolve(undefined); + const mapPromise = plan.map + ? trpc.world.getMap.query({ generalId: id, showMe: true, useCache: true }) + : Promise.resolve(undefined); + const commandsPromise = plan.commands + ? trpc.turns.getCommandTable.query({ generalId: id }) + : Promise.resolve(undefined); + const contactsPromise = plan.contacts + ? trpc.messages.getContacts.query({ generalId: id }) + : Promise.resolve(undefined); + const boardPromise = plan.boardAccess ? trpc.board.getAccess.query() : Promise.resolve(undefined); + const reservedPromise = plan.reservedTurns + ? trpc.turns.reserved.getGeneral.query({ generalId: id }) + : Promise.resolve(undefined); + const recordsPromise = plan.records + ? trpc.general.getRecentRecords + .query({ lastGeneralRecordId, lastWorldHistoryId }) + .catch((err: unknown) => { + recordsError.value = resolveErrorMessage(err); + return null; + }) + : Promise.resolve(undefined); + const frontPromise = plan.frontStatus + ? trpc.general.getFrontStatus.query().catch((err: unknown) => { + frontStatusError.value = resolveErrorMessage(err); + return null; + }) + : Promise.resolve(undefined); + + const [context, lobby, map, commands, contacts, access, generalTurns, records, nextFrontStatus] = + await Promise.all([ + contextPromise, + lobbyPromise, + mapPromise, + commandsPromise, + contactsPromise, + boardPromise, + reservedPromise, + recordsPromise, + frontPromise, + ]); + + if (context === null) { + general.value = null; + city.value = null; + nation.value = null; + reservedGeneralTurns.value = null; + reservedGeneralRevision.value = 0; + boardAccess.value = null; + resetRecentRecords(null); + return; + } + if (context !== undefined) { + general.value = structurallyShare(general.value, context.general); + city.value = structurallyShare(city.value, context.city); + nation.value = structurallyShare(nation.value, context.nation); + } + if (lobby !== undefined) lobbyInfo.value = structurallyShare(lobbyInfo.value, lobby); + if (map !== undefined) worldMap.value = structurallyShare(worldMap.value, map); + if (commands !== undefined) commandTable.value = structurallyShare(commandTable.value, commands); + if (contacts !== undefined) messageContacts.value = structurallyShare(messageContacts.value, contacts); + if (access !== undefined) boardAccess.value = structurallyShare(boardAccess.value, access); + if (generalTurns !== undefined) { + reservedGeneralTurns.value = structurallyShare( + reservedGeneralTurns.value, + generalTurns.turns + ) as ReservedTurnView[]; + reservedGeneralRevision.value = generalTurns.revision; + } + if (records) applyRecentRecords(records); + if (nextFrontStatus) updateFrontStatus(nextFrontStatus); + } catch (err) { + error.value = resolveErrorMessage(err); + } finally { + refreshing.value = false; + } + }; + + const readModelRefreshQueue = createMergedReadModelRefreshQueue(refreshChangedReadModels); + const refreshMessages = async () => { const id = generalId.value; if (!id) { @@ -701,8 +810,24 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => { source.addEventListener('error', () => { realtimeStatus.value = realtimeEnabled.value ? 'idle' : 'paused'; }); - source.addEventListener('turnCompleted', () => { - realtimeRefreshQueue.request(); + source.addEventListener('turnCompleted', (event) => { + const payload = parseRealtimePayload(event); + if (!payload || payload.type !== 'turnCompleted') { + return; + } + if (!payload.changes) { + // Rolling deployment fallback for an older daemon. + realtimeRefreshQueue.request(); + return; + } + readModelRefreshQueue.request(payload.changes); + }); + source.addEventListener('readModelChanged', (event) => { + const payload = parseRealtimePayload(event); + if (!payload || payload.type !== 'readModelChanged') { + return; + } + readModelRefreshQueue.request(payload.changes); }); source.addEventListener('messageCreated', (event) => { const payload = parseRealtimePayload(event); @@ -724,6 +849,7 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => { if (!realtimeActive.value) return; if (document.visibilityState === 'hidden') { realtimeRefreshQueue.cancelPending(); + readModelRefreshQueue.cancelPending(); closeRealtimeSource(); realtimeStatus.value = 'idle'; return; @@ -746,6 +872,7 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => { const stopRealtime = () => { realtimeActive.value = false; realtimeRefreshQueue.cancelPending(); + readModelRefreshQueue.cancelPending(); closeRealtimeSource(); if (visibilityListenerInstalled) { document.removeEventListener('visibilitychange', handleVisibilityChange); diff --git a/app/game-frontend/src/utils/dashboardReadModel.ts b/app/game-frontend/src/utils/dashboardReadModel.ts new file mode 100644 index 00000000..63bf4d16 --- /dev/null +++ b/app/game-frontend/src/utils/dashboardReadModel.ts @@ -0,0 +1,118 @@ +import { + createEmptyRealtimeReadModelChanges, + mergeRealtimeReadModelChanges, + type RealtimeReadModelChanges, +} from '@sammo-ts/common'; + +export interface DashboardReadModelIdentity { + generalId: number | null; + cityId: number | null; + nationId: number | null; +} + +export interface DashboardRefreshPlan { + context: boolean; + lobby: boolean; + map: boolean; + commands: boolean; + contacts: boolean; + boardAccess: boolean; + reservedTurns: boolean; + records: boolean; + frontStatus: boolean; +} + +const contains = (ids: readonly number[], id: number | null): boolean => id !== null && ids.includes(id); + +export const resolveDashboardRefreshPlan = ( + changes: RealtimeReadModelChanges, + identity: DashboardReadModelIdentity +): DashboardRefreshPlan => { + const ownGeneralChanged = contains(changes.generalIds, identity.generalId); + const ownCityChanged = contains(changes.cityIds, identity.cityId); + const ownNationChanged = contains(changes.nationIds, identity.nationId); + const entityContextChanged = ownGeneralChanged || ownCityChanged || ownNationChanged; + const worldEntitiesChanged = changes.cityIds.length > 0 || changes.nationIds.length > 0; + + return { + context: entityContextChanged, + lobby: changes.worldChanged || changes.contactsChanged, + map: changes.worldChanged || worldEntitiesChanged || ownGeneralChanged, + commands: changes.worldChanged || worldEntitiesChanged || ownGeneralChanged, + contacts: changes.contactsChanged, + boardAccess: entityContextChanged, + reservedTurns: contains(changes.reservedGeneralIds, identity.generalId), + records: + changes.globalRecordsChanged || + changes.worldHistoryChanged || + contains(changes.recordGeneralIds, identity.generalId), + // lastTurnTime is intentionally excluded. This slice contains the + // nation notice/vote/presence model and only follows related changes. + frontStatus: changes.contactsChanged || ownNationChanged, + }; +}; + +type TimerHandle = ReturnType; + +export interface MergedReadModelRefreshQueue { + request(changes: RealtimeReadModelChanges): void; + cancelPending(): void; +} + +export const createMergedReadModelRefreshQueue = ( + refresh: (changes: RealtimeReadModelChanges) => Promise, + options: { + minIntervalMs?: number; + now?: () => number; + setTimer?: (callback: () => void, delayMs: number) => TimerHandle; + clearTimer?: (handle: TimerHandle) => void; + } = {} +): MergedReadModelRefreshQueue => { + const minIntervalMs = Math.max(0, options.minIntervalMs ?? 1_000); + const now = options.now ?? Date.now; + const setTimer = options.setTimer ?? ((callback, delayMs) => setTimeout(callback, delayMs)); + const clearTimer = options.clearTimer ?? ((handle) => clearTimeout(handle)); + let pending = createEmptyRealtimeReadModelChanges(); + let hasPending = false; + let running = false; + let timer: TimerHandle | null = null; + let lastStartedAt = Number.NEGATIVE_INFINITY; + + const schedule = () => { + if (!hasPending || running || timer !== null) { + return; + } + const delayMs = Math.max(0, lastStartedAt + minIntervalMs - now()); + timer = setTimer(() => { + timer = null; + if (!hasPending || running) { + return; + } + const next = pending; + pending = createEmptyRealtimeReadModelChanges(); + hasPending = false; + running = true; + lastStartedAt = now(); + void refresh(next).finally(() => { + running = false; + schedule(); + }); + }, delayMs); + }; + + return { + request: (changes) => { + pending = hasPending ? mergeRealtimeReadModelChanges(pending, changes) : changes; + hasPending = true; + schedule(); + }, + cancelPending: () => { + hasPending = false; + pending = createEmptyRealtimeReadModelChanges(); + if (timer !== null) { + clearTimer(timer); + timer = null; + } + }, + }; +}; diff --git a/app/game-frontend/test/dashboardReadModel.test.ts b/app/game-frontend/test/dashboardReadModel.test.ts new file mode 100644 index 00000000..c4a09999 --- /dev/null +++ b/app/game-frontend/test/dashboardReadModel.test.ts @@ -0,0 +1,94 @@ +import assert from 'node:assert/strict'; +import test from 'node:test'; + +import { createEmptyRealtimeReadModelChanges } from '@sammo-ts/common'; +import { + createMergedReadModelRefreshQueue, + resolveDashboardRefreshPlan, +} from '../src/utils/dashboardReadModel.ts'; + +void test('last-turn-time-only events do not schedule any dashboard query', () => { + const plan = resolveDashboardRefreshPlan(createEmptyRealtimeReadModelChanges(), { + generalId: 7, + cityId: 3, + nationId: 2, + }); + + assert.deepEqual(plan, { + context: false, + lobby: false, + map: false, + commands: false, + contacts: false, + boardAccess: false, + reservedTurns: false, + records: false, + frontStatus: false, + }); +}); + +void test('selects only the read models affected by the current identity', () => { + const changes = { + ...createEmptyRealtimeReadModelChanges(), + generalIds: [7, 99], + reservedGeneralIds: [7], + recordGeneralIds: [7], + }; + + assert.deepEqual(resolveDashboardRefreshPlan(changes, { generalId: 7, cityId: 3, nationId: 2 }), { + context: true, + lobby: false, + map: true, + commands: true, + contacts: false, + boardAccess: true, + reservedTurns: true, + records: true, + frontStatus: false, + }); +}); + +void test('merges burst payloads without losing entity ids and starts at most once per interval', async () => { + let nowMs = 0; + let nextTimerId = 1; + const timers = new Map void; at: number }>(); + const observed: number[][] = []; + const queue = createMergedReadModelRefreshQueue( + async (changes) => { + observed.push(changes.generalIds); + }, + { + minIntervalMs: 1_000, + now: () => nowMs, + setTimer: (callback, delayMs) => { + const id = nextTimerId++; + timers.set(id, { callback, at: nowMs + delayMs }); + return id as unknown as ReturnType; + }, + clearTimer: (timer) => timers.delete(timer as unknown as number), + } + ); + const runDueTimers = () => { + for (const [id, timer] of [...timers]) { + if (timer.at <= nowMs) { + timers.delete(id); + timer.callback(); + } + } + }; + + queue.request({ ...createEmptyRealtimeReadModelChanges(), generalIds: [7] }); + runDueTimers(); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(observed, [[7]]); + + queue.request({ ...createEmptyRealtimeReadModelChanges(), generalIds: [9] }); + queue.request({ ...createEmptyRealtimeReadModelChanges(), generalIds: [8, 9] }); + nowMs = 999; + runDueTimers(); + assert.equal(observed.length, 1); + nowMs = 1_000; + runDueTimers(); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(observed, [[7], [8, 9]]); +}); diff --git a/packages/common/src/realtime/keys.ts b/packages/common/src/realtime/keys.ts index 2b638cc7..6fc1baff 100644 --- a/packages/common/src/realtime/keys.ts +++ b/packages/common/src/realtime/keys.ts @@ -7,3 +7,13 @@ export const buildGameEventChannel = (profileName: string): string => { const normalized = normalizeProfileName(profileName); return `sammo:${normalized}:realtime:events`; }; + +export const buildGameReadModelRevisionKey = (profileName: string): string => { + const normalized = normalizeProfileName(profileName); + return `sammo:${normalized}:read-model:revision`; +}; + +export const buildGameReadModelDomainRevisionKey = (profileName: string): string => { + const normalized = normalizeProfileName(profileName); + return `sammo:${normalized}:read-model:domain-revisions`; +}; diff --git a/packages/common/src/realtime/types.ts b/packages/common/src/realtime/types.ts index d873d4bf..4fe201ed 100644 --- a/packages/common/src/realtime/types.ts +++ b/packages/common/src/realtime/types.ts @@ -1,9 +1,77 @@ export type MessageTypeKey = 'public' | 'private' | 'national' | 'diplomacy'; +/** + * Durable mutations summarized after the database transaction commits. + * Entity IDs let each authenticated client decide whether its own read model + * is affected without exposing entity payloads over the shared Redis channel. + */ +export interface RealtimeReadModelChanges { + generalIds: number[]; + cityIds: number[]; + nationIds: number[]; + reservedGeneralIds: number[]; + recordGeneralIds: number[]; + worldChanged: boolean; + globalRecordsChanged: boolean; + worldHistoryChanged: boolean; + contactsChanged: boolean; +} + +export const createEmptyRealtimeReadModelChanges = (): RealtimeReadModelChanges => ({ + generalIds: [], + cityIds: [], + nationIds: [], + reservedGeneralIds: [], + recordGeneralIds: [], + worldChanged: false, + globalRecordsChanged: false, + worldHistoryChanged: false, + contactsChanged: false, +}); + +const mergeIds = (left: readonly number[], right: readonly number[]): number[] => + [...new Set([...left, ...right])].sort((a, b) => a - b); + +export const mergeRealtimeReadModelChanges = ( + left: RealtimeReadModelChanges, + right: RealtimeReadModelChanges +): RealtimeReadModelChanges => ({ + generalIds: mergeIds(left.generalIds, right.generalIds), + cityIds: mergeIds(left.cityIds, right.cityIds), + nationIds: mergeIds(left.nationIds, right.nationIds), + reservedGeneralIds: mergeIds(left.reservedGeneralIds, right.reservedGeneralIds), + recordGeneralIds: mergeIds(left.recordGeneralIds, right.recordGeneralIds), + worldChanged: left.worldChanged || right.worldChanged, + globalRecordsChanged: left.globalRecordsChanged || right.globalRecordsChanged, + worldHistoryChanged: left.worldHistoryChanged || right.worldHistoryChanged, + contactsChanged: left.contactsChanged || right.contactsChanged, +}); + +export const hasRealtimeReadModelChanges = (changes: RealtimeReadModelChanges): boolean => + changes.generalIds.length > 0 || + changes.cityIds.length > 0 || + changes.nationIds.length > 0 || + changes.reservedGeneralIds.length > 0 || + changes.recordGeneralIds.length > 0 || + changes.worldChanged || + changes.globalRecordsChanged || + changes.worldHistoryChanged || + changes.contactsChanged; + export interface TurnCompletedEvent { type: 'turnCompleted'; at: string; lastTurnTime: string; + /** Absent only for a rolling-deploy event produced by an older daemon. */ + changes?: RealtimeReadModelChanges; + revision?: number; +} + +export interface ReadModelChangedEvent { + type: 'readModelChanged'; + at: string; + changes: RealtimeReadModelChanges; + revision: number; } export interface MessageCreatedEvent { @@ -17,4 +85,5 @@ export interface MessageCreatedEvent { export type RealtimeEvent = | TurnCompletedEvent + | ReadModelChangedEvent | MessageCreatedEvent;