refactor(game): refresh committed read models selectively

This commit is contained in:
2026-08-09 15:46:01 +00:00
parent 00b9fcdb93
commit a2683df80f
14 changed files with 807 additions and 48 deletions
+20 -2
View File
@@ -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<string, unknown>): Record<number, number> =
const buildBaseMapCacheKey = (ctx: GameApiContext, scope: 'base' | 'public' = 'base'): string =>
`sammo:map:${scope}:${ctx.profile.id}:${ctx.profile.scenario}`;
const loadWorldMapRevision = async (ctx: GameApiContext): Promise<string> => {
const redis = ctx.redis as unknown as {
hGet?: (key: string, field: string) => Promise<string | null>;
};
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<string> =>
`${buildBaseMapCacheKey(ctx)}:r${await loadWorldMapRevision(ctx)}`;
const loadBaseMap = async (
ctx: GameApiContext,
options?: {
@@ -103,7 +121,7 @@ const loadBaseMap = async (
}
): Promise<BaseMapResult | null> => {
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) {
@@ -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'
);
});
});
@@ -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);
}
+1
View File
@@ -70,6 +70,7 @@ export interface TurnDaemonHooks {
requestId: string,
execute: (context: TurnDaemonCommandExecutionContext) => Promise<TurnDaemonCommandResult>
): Promise<TurnDaemonCommandResult>;
publishCommandEvents?(result: TurnDaemonCommandResult): Promise<void>;
publishEvents?(result: TurnRunResult): Promise<void>;
onRunError?(error: unknown): Promise<void>;
}
+103 -15
View File
@@ -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<void>;
}
const uniqueSortedIds = (values: Iterable<number>): 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(),
};
};
+56 -2
View File
@@ -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<void>) | null = null;
let publishReadModelChanges: ((changes: RealtimeReadModelChanges) => Promise<number>) | null = null;
let takeCommittedReadModelChanges: (() => RealtimeReadModelChanges | null) | null = null;
let close = async () => {};
let auctionFinalizer: Awaited<ReturnType<typeof createAuctionFinalizer>> | null = null;
let auctionBidder: Awaited<ReturnType<typeof createAuctionBidder>> | 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<number | undefined> => {
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);
},
};
}
@@ -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;
@@ -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,
});
});
});
+72 -7
View File
@@ -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));
+148 -21
View File
@@ -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<ReturnType<typeof trpc.general.getRecentRecords.query>>
) => {
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<unknown>(
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);
@@ -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<typeof setTimeout>;
export interface MergedReadModelRefreshQueue {
request(changes: RealtimeReadModelChanges): void;
cancelPending(): void;
}
export const createMergedReadModelRefreshQueue = (
refresh: (changes: RealtimeReadModelChanges) => Promise<void>,
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;
}
},
};
};
@@ -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<number, { callback: () => 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<typeof setTimeout>;
},
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<void>((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<void>((resolve) => setImmediate(resolve));
assert.deepEqual(observed, [[7], [8, 9]]);
});