merge: coalesce dashboard realtime across tabs

This commit is contained in:
2026-08-09 23:55:14 +00:00
8 changed files with 676 additions and 68 deletions
@@ -306,6 +306,7 @@ integration('adjustGeneralIcon PostgreSQL persistence', () => {
CHECK (request_id <> '${rollbackRequestId}' OR status <> 'SUCCEEDED') CHECK (request_id <> '${rollbackRequestId}' OR status <> 'SUCCEEDED')
`); `);
await expect(execute(rollbackCommand)).rejects.toThrow(`violates check constraint "${rollbackConstraint}"`); await expect(execute(rollbackCommand)).rejects.toThrow(`violates check constraint "${rollbackConstraint}"`);
expect(hooks.takeCommittedReadModelChanges()).toBeNull();
expect(world.getGeneralById(targetGeneralId)).toMatchObject({ expect(world.getGeneralById(targetGeneralId)).toMatchObject({
picture: initialPicture, picture: initialPicture,
imageServer: initialImageServer, imageServer: initialImageServer,
@@ -334,6 +335,7 @@ integration('adjustGeneralIcon PostgreSQL persistence', () => {
); );
await createInputEvent(actorMismatchCommand, foreignUserId); await createInputEvent(actorMismatchCommand, foreignUserId);
await expect(execute(actorMismatchCommand)).rejects.toThrow('actor does not match'); await expect(execute(actorMismatchCommand)).rejects.toThrow('actor does not match');
expect(hooks.takeCommittedReadModelChanges()).toBeNull();
expect(world.getGeneralById(targetGeneralId)).toMatchObject({ expect(world.getGeneralById(targetGeneralId)).toMatchObject({
picture: initialPicture, picture: initialPicture,
imageServer: initialImageServer, imageServer: initialImageServer,
@@ -357,6 +359,13 @@ integration('adjustGeneralIcon PostgreSQL persistence', () => {
generalId: targetGeneralId, generalId: targetGeneralId,
updated: true, updated: true,
}); });
expect(hooks.takeCommittedReadModelChanges()).toMatchObject({
generalIds: [targetGeneralId],
mapGeneralIds: [],
frontStatusGeneralIds: [],
lobbyGeneralIds: [targetGeneralId],
reservedGeneralIds: [],
});
expect(world.getGeneralById(targetGeneralId)).toMatchObject({ expect(world.getGeneralById(targetGeneralId)).toMatchObject({
picture: nextPicture, picture: nextPicture,
imageServer: nextImageServer, imageServer: nextImageServer,
+164 -9
View File
@@ -216,17 +216,23 @@ const installRealtimeHarness = async (page: Page) => {
await page.addInitScript(() => { await page.addInitScript(() => {
class TestEventSource extends EventTarget { class TestEventSource extends EventTarget {
static latest: TestEventSource | null = null; static latest: TestEventSource | null = null;
static created = 0;
static closed = 0;
readonly url: string; readonly url: string;
constructor(url: string | URL) { constructor(url: string | URL) {
super(); super();
this.url = url.toString(); this.url = url.toString();
TestEventSource.created += 1;
TestEventSource.latest = this; TestEventSource.latest = this;
queueMicrotask(() => this.dispatchEvent(new Event('open'))); queueMicrotask(() => this.dispatchEvent(new Event('open')));
} }
close() { close() {
if (TestEventSource.latest === this) TestEventSource.latest = null; if (TestEventSource.latest === this) {
TestEventSource.latest = null;
TestEventSource.closed += 1;
}
} }
} }
@@ -243,6 +249,14 @@ const installRealtimeHarness = async (page: Page) => {
configurable: true, configurable: true,
value: () => TestEventSource.latest !== null, value: () => TestEventSource.latest !== null,
}); });
Object.defineProperty(window, '__mainRealtimeStats', {
configurable: true,
value: () => ({
active: TestEventSource.latest !== null,
created: TestEventSource.created,
closed: TestEventSource.closed,
}),
});
}); });
}; };
@@ -517,7 +531,9 @@ test('mobile single document refreshes once and preserves tokens on lobby return
expect(state.operations).not.toContain('auth.logout'); expect(state.operations).not.toContain('auth.logout');
}); });
test('realtime read-model events skip clock-only work, merge bursts, patch in place, and stop off-route', 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 = { const state: NavigationFixture = {
officerLevel: 5, officerLevel: 5,
permission: 2, permission: 2,
@@ -673,9 +689,9 @@ test('realtime read-model events skip clock-only work, merge bursts, patch in pl
} }
); );
}); });
await expect.poll(() => state.operations.slice(operationsBeforeSurvey), { timeout: 3_000 }).toEqual([ await expect
'general.getFrontStatus', .poll(() => state.operations.slice(operationsBeforeSurvey), { timeout: 3_000 })
]); .toEqual(['general.getFrontStatus']);
const profile = await page.evaluate(() => { const profile = await page.evaluate(() => {
const probe = ( const probe = (
@@ -725,10 +741,7 @@ test('realtime read-model events skip clock-only work, merge bursts, patch in pl
await page.locator(`a[href="${basePath}/board"]`).first().click(); await page.locator(`a[href="${basePath}/board"]`).first().click();
await page.waitForURL(`**${basePath}/board`); await page.waitForURL(`**${basePath}/board`);
expect( expect(
await page.evaluate( await page.evaluate(() => (window as unknown as { __hasMainRealtime: () => boolean }).__hasMainRealtime())
() =>
(window as unknown as { __hasMainRealtime: () => boolean }).__hasMainRealtime()
)
).toBe(false); ).toBe(false);
const callsAfterLeavingMain = state.generalMeCalls; const callsAfterLeavingMain = state.generalMeCalls;
await page.evaluate(() => { await page.evaluate(() => {
@@ -763,3 +776,145 @@ test('realtime read-model events skip clock-only work, merge bursts, patch in pl
await new Promise((resolve) => setTimeout(resolve, 300)); await new Promise((resolve) => setTimeout(resolve, 300));
expect(state.generalMeCalls).toBe(callsAfterLeavingMain); expect(state.generalMeCalls).toBe(callsAfterLeavingMain);
}); });
test('same-account main tabs share one realtime diff and exclude a tab while sync is off', async ({
context,
page,
}) => {
const state: NavigationFixture = {
officerLevel: 5,
permission: 2,
nationLevel: 3,
stage: 0,
npcMode: 1,
generalMeCalls: 0,
operations: [],
};
const secondPage = await context.newPage();
const pages = [page, secondPage];
for (const currentPage of pages) {
await currentPage.addInitScript(() => {
Object.defineProperty(document, 'visibilityState', { configurable: true, get: () => 'visible' });
});
await installRealtimeHarness(currentPage);
await installFixture(currentPage, state);
await currentPage.setViewportSize({ width: 1200, height: 900 });
}
await Promise.all(pages.map((currentPage) => waitForMain(currentPage)));
await expect
.poll(async () => {
const stats = await Promise.all(
pages.map((currentPage) =>
currentPage.evaluate(() =>
(
window as unknown as {
__mainRealtimeStats: () => { active: boolean; created: number; closed: number };
}
).__mainRealtimeStats()
)
)
);
return stats.filter((entry) => entry.active).length;
})
.toBe(1);
const activeFlags = await Promise.all(
pages.map((currentPage) =>
currentPage.evaluate(() => (window as unknown as { __hasMainRealtime: () => boolean }).__hasMainRealtime())
)
);
const leaderIndex = activeFlags.findIndex(Boolean);
const followerIndex = leaderIndex === 0 ? 1 : 0;
const leaderPage = pages[leaderIndex];
const followerPage = pages[followerIndex];
if (!leaderPage || !followerPage) throw new Error('realtime leader election failed');
const callsBeforeSharedRefresh = state.generalMeCalls;
state.generalName = '탭공유갱신장수';
await leaderPage.evaluate(() => {
(window as unknown as { __emitMainRealtime: (type: string, payload: unknown) => void }).__emitMainRealtime(
'readModelChanged',
{
at: new Date().toISOString(),
revision: 100,
changes: {
generalIds: [7],
cityIds: [],
nationIds: [],
mapGeneralIds: [],
mapCityIds: [],
mapNationIds: [],
frontStatusGeneralIds: [],
frontStatusNationIds: [],
frontStatusActorIds: [],
frontStatusChanged: false,
lobbyGeneralIds: [],
lobbyChanged: false,
reservedGeneralIds: [],
recordGeneralIds: [],
worldChanged: false,
globalRecordsChanged: false,
worldHistoryChanged: false,
contactsChanged: false,
},
}
);
});
await expect.poll(() => state.generalMeCalls).toBe(callsBeforeSharedRefresh + 1);
await Promise.all(
pages.map((currentPage) => expect(currentPage.locator('.general-title')).toContainText('탭공유갱신장수'))
);
await followerPage.getByRole('button', { name: / /u }).click();
await expect(followerPage.getByRole('button', { name: / : /u })).toBeVisible();
const callsBeforeExcludedRefresh = state.generalMeCalls;
state.generalName = '리더만갱신장수';
await leaderPage.evaluate(() => {
(window as unknown as { __emitMainRealtime: (type: string, payload: unknown) => void }).__emitMainRealtime(
'readModelChanged',
{
at: new Date().toISOString(),
revision: 101,
changes: {
generalIds: [7],
cityIds: [],
nationIds: [],
mapGeneralIds: [],
mapCityIds: [],
mapNationIds: [],
frontStatusGeneralIds: [],
frontStatusNationIds: [],
frontStatusActorIds: [],
frontStatusChanged: false,
lobbyGeneralIds: [],
lobbyChanged: false,
reservedGeneralIds: [],
recordGeneralIds: [],
worldChanged: false,
globalRecordsChanged: false,
worldHistoryChanged: false,
contactsChanged: false,
},
}
);
});
await expect.poll(() => state.generalMeCalls).toBe(callsBeforeExcludedRefresh + 1);
await expect(leaderPage.locator('.general-title')).toContainText('리더만갱신장수');
await expect(followerPage.locator('.general-title')).toContainText('탭공유갱신장수');
await followerPage.getByRole('button', { name: / : /u }).click();
await expect(followerPage.locator('.general-title')).toContainText('리더만갱신장수');
await expect
.poll(async () => {
const flags = await Promise.all(
pages.map((currentPage) =>
currentPage.evaluate(() =>
(window as unknown as { __hasMainRealtime: () => boolean }).__hasMainRealtime()
)
)
);
return flags.filter(Boolean).length;
})
.toBe(1);
});
+233 -58
View File
@@ -1,4 +1,4 @@
import { computed, ref, watch } from 'vue'; import { computed, ref, toRaw, watch } from 'vue';
import { defineStore } from 'pinia'; import { defineStore } from 'pinia';
import { MESSAGE_MAILBOX_NATIONAL_BASE, MESSAGE_MAILBOX_PUBLIC, type MessageType } from '@sammo-ts/logic'; import { MESSAGE_MAILBOX_NATIONAL_BASE, MESSAGE_MAILBOX_PUBLIC, type MessageType } from '@sammo-ts/logic';
import type { RealtimeEvent, RealtimeReadModelChanges } from '@sammo-ts/common'; import type { RealtimeEvent, RealtimeReadModelChanges } from '@sammo-ts/common';
@@ -9,6 +9,7 @@ import { createLatestRefreshQueue } from '../utils/latestRefreshQueue';
import { createRateLimitedRefreshQueue } from '../utils/rateLimitedRefreshQueue'; import { createRateLimitedRefreshQueue } from '../utils/rateLimitedRefreshQueue';
import { structurallyShare } from '../utils/structuralShare'; import { structurallyShare } from '../utils/structuralShare';
import { createMergedReadModelRefreshQueue, resolveDashboardRefreshPlan } from '../utils/dashboardReadModel'; import { createMergedReadModelRefreshQueue, resolveDashboardRefreshPlan } from '../utils/dashboardReadModel';
import { createBroadcastTabCoordinator, type BroadcastTabCoordinator } from '../utils/broadcastTabCoordinator';
const REALTIME_FULL_REFRESH_MIN_INTERVAL_MS = 5_000; const REALTIME_FULL_REFRESH_MIN_INTERVAL_MS = 5_000;
@@ -35,6 +36,26 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
type ReservedTurnView = Awaited<ReturnType<typeof trpc.turns.reserved.getGeneral.query>>['turns'][number]; type ReservedTurnView = Awaited<ReturnType<typeof trpc.turns.reserved.getGeneral.query>>['turns'][number];
type RecentRecord = Awaited<ReturnType<typeof trpc.general.getRecentRecords.query>>['global'][number]; type RecentRecord = Awaited<ReturnType<typeof trpc.general.getRecentRecords.query>>['global'][number];
type FrontStatus = Awaited<ReturnType<typeof trpc.general.getFrontStatus.query>>; type FrontStatus = Awaited<ReturnType<typeof trpc.general.getFrontStatus.query>>;
type DashboardReadModelPatch = {
general?: PresentGeneralContext['general'] | null;
city?: PresentGeneralContext['city'] | null;
nation?: PresentGeneralContext['nation'] | null;
lobbyInfo?: LobbyInfo | null;
worldMap?: WorldMapResult | null;
mapLayout?: MapLayout | null;
commandTable?: CommandTable | null;
messages?: MessageBundle | null;
messageContacts?: MessageContacts | null;
boardAccess?: BoardAccess | null;
reservedGeneralTurns?: ReservedTurnView[] | null;
reservedGeneralRevision?: number;
globalRecords?: RecentRecord[];
generalRecords?: RecentRecord[];
worldHistory?: RecentRecord[];
frontStatus?: FrontStatus | null;
};
type DashboardTabMessage =
{ kind: 'patch'; patch: DashboardReadModelPatch } | { kind: 'status'; status: 'idle' | 'connected' };
const loading = ref(false); const loading = ref(false);
const refreshing = ref(false); const refreshing = ref(false);
@@ -206,9 +227,15 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
}); });
const setRealtimeEnabled = (enabled: boolean) => { const setRealtimeEnabled = (enabled: boolean) => {
const wasEnabled = realtimeEnabled.value;
realtimeEnabled.value = enabled; realtimeEnabled.value = enabled;
if (!enabled) { if (!enabled) {
realtimeStatus.value = 'paused'; realtimeStatus.value = 'paused';
reconcileRealtimeCoordinator();
return;
}
if (!wasEnabled) {
void refreshQueue.request().finally(() => reconcileRealtimeCoordinator());
} }
}; };
@@ -253,9 +280,7 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
surveyNotice.value = null; surveyNotice.value = null;
}; };
const applyRecentRecords = ( const applyRecentRecords = (records: Awaited<ReturnType<typeof trpc.general.getRecentRecords.query>>) => {
records: Awaited<ReturnType<typeof trpc.general.getRecentRecords.query>>
) => {
globalRecords.value = structurallyShare( globalRecords.value = structurallyShare(
globalRecords.value, globalRecords.value,
mergeRecentRecords(globalRecords.value, records.global) mergeRecentRecords(globalRecords.value, records.global)
@@ -268,14 +293,86 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
worldHistory.value, worldHistory.value,
mergeRecentRecords(worldHistory.value, records.history) mergeRecentRecords(worldHistory.value, records.history)
); );
lastGeneralRecordId = Math.max( lastGeneralRecordId = Math.max(lastGeneralRecordId, records.global[0]?.id ?? 0, records.general[0]?.id ?? 0);
lastGeneralRecordId,
records.global[0]?.id ?? 0,
records.general[0]?.id ?? 0
);
lastWorldHistoryId = Math.max(lastWorldHistoryId, records.history[0]?.id ?? 0); lastWorldHistoryId = Math.max(lastWorldHistoryId, records.history[0]?.id ?? 0);
}; };
const applyDashboardPatch = (patch: DashboardReadModelPatch) => {
if (patch.general === null) {
general.value = null;
city.value = null;
nation.value = null;
reservedGeneralTurns.value = null;
reservedGeneralRevision.value = 0;
boardAccess.value = null;
resetRecentRecords(null);
} else if (patch.general !== undefined) {
general.value = structurallyShare(general.value, patch.general);
}
if (patch.city !== undefined) city.value = structurallyShare(city.value, patch.city);
if (patch.nation !== undefined) nation.value = structurallyShare(nation.value, patch.nation);
if (patch.lobbyInfo !== undefined) lobbyInfo.value = structurallyShare(lobbyInfo.value, patch.lobbyInfo);
if (patch.worldMap !== undefined) worldMap.value = structurallyShare(worldMap.value, patch.worldMap);
if (patch.mapLayout !== undefined) mapLayout.value = structurallyShare(mapLayout.value, patch.mapLayout);
if (patch.commandTable !== undefined) {
commandTable.value = structurallyShare(commandTable.value, patch.commandTable);
}
if (patch.messages !== undefined) messages.value = structurallyShare(messages.value, patch.messages);
if (patch.messageContacts !== undefined) {
messageContacts.value = structurallyShare(messageContacts.value, patch.messageContacts);
}
if (patch.boardAccess !== undefined)
boardAccess.value = structurallyShare(boardAccess.value, patch.boardAccess);
if (patch.reservedGeneralTurns !== undefined) {
reservedGeneralTurns.value = structurallyShare<unknown>(
reservedGeneralTurns.value,
patch.reservedGeneralTurns
) as ReservedTurnView[] | null;
}
if (patch.reservedGeneralRevision !== undefined) {
reservedGeneralRevision.value = patch.reservedGeneralRevision;
}
if (patch.globalRecords !== undefined) {
globalRecords.value = structurallyShare(globalRecords.value, patch.globalRecords);
lastGeneralRecordId = Math.max(lastGeneralRecordId, patch.globalRecords[0]?.id ?? 0);
}
if (patch.generalRecords !== undefined) {
generalRecords.value = structurallyShare(generalRecords.value, patch.generalRecords);
lastGeneralRecordId = Math.max(lastGeneralRecordId, patch.generalRecords[0]?.id ?? 0);
}
if (patch.worldHistory !== undefined) {
worldHistory.value = structurallyShare(worldHistory.value, patch.worldHistory);
lastWorldHistoryId = Math.max(lastWorldHistoryId, patch.worldHistory[0]?.id ?? 0);
}
if (patch.frontStatus === null) {
frontStatus.value = null;
surveyNotice.value = null;
} else if (patch.frontStatus !== undefined) {
updateFrontStatus(patch.frontStatus);
}
};
const currentDashboardPatch = (): DashboardReadModelPatch => {
const patch: DashboardReadModelPatch = {};
patch.general = toRaw(general.value);
patch.city = toRaw(city.value);
patch.nation = toRaw(nation.value);
patch.lobbyInfo = toRaw(lobbyInfo.value);
patch.worldMap = toRaw(worldMap.value);
patch.mapLayout = toRaw(mapLayout.value);
patch.commandTable = toRaw(commandTable.value);
patch.messages = toRaw(messages.value);
patch.messageContacts = toRaw(messageContacts.value);
patch.boardAccess = toRaw(boardAccess.value);
patch.reservedGeneralTurns = toRaw(reservedGeneralTurns.value as unknown) as ReservedTurnView[] | null;
patch.reservedGeneralRevision = reservedGeneralRevision.value;
patch.globalRecords = toRaw(globalRecords.value);
patch.generalRecords = toRaw(generalRecords.value);
patch.worldHistory = toRaw(worldHistory.value);
patch.frontStatus = toRaw(frontStatus.value);
return patch;
};
const refreshMainData = async () => { const refreshMainData = async () => {
const isInitialLoad = !initialized; const isInitialLoad = !initialized;
if (isInitialLoad) { if (isInitialLoad) {
@@ -384,9 +481,22 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
const refreshQueue = createLatestRefreshQueue(refreshMainData); const refreshQueue = createLatestRefreshQueue(refreshMainData);
const loadMainData = () => refreshQueue.request(); const loadMainData = () => refreshQueue.request();
const realtimeRefreshQueue = createRateLimitedRefreshQueue(() => refreshQueue.request(), { let realtimeCoordinator: BroadcastTabCoordinator<DashboardTabMessage> | null = null;
minIntervalMs: REALTIME_FULL_REFRESH_MIN_INTERVAL_MS, let realtimeCoordinatorScope: string | null = null;
});
const publishDashboardPatch = (patch: DashboardReadModelPatch) => {
realtimeCoordinator?.postFromLeader({ kind: 'patch', patch });
};
const realtimeRefreshQueue = createRateLimitedRefreshQueue(
async () => {
await refreshQueue.request();
publishDashboardPatch(currentDashboardPatch());
},
{
minIntervalMs: REALTIME_FULL_REFRESH_MIN_INTERVAL_MS,
}
);
const refreshChangedReadModels = async (changes: RealtimeReadModelChanges) => { const refreshChangedReadModels = async (changes: RealtimeReadModelChanges) => {
const id = generalId.value; const id = generalId.value;
@@ -452,35 +562,34 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
frontPromise, frontPromise,
]); ]);
const patch: DashboardReadModelPatch = {};
if (context === null) { if (context === null) {
general.value = null; patch.general = null;
city.value = null; } else if (context !== undefined) {
nation.value = null; patch.general = context.general;
reservedGeneralTurns.value = null; patch.city = context.city;
reservedGeneralRevision.value = 0; patch.nation = context.nation;
boardAccess.value = null;
resetRecentRecords(null);
return;
} }
if (context !== undefined) { if (lobby !== undefined) patch.lobbyInfo = lobby;
general.value = structurallyShare(general.value, context.general); if (map !== undefined) patch.worldMap = map;
city.value = structurallyShare(city.value, context.city); if (commands !== undefined) patch.commandTable = commands;
nation.value = structurallyShare(nation.value, context.nation); if (contacts !== undefined) patch.messageContacts = contacts;
} if (access !== undefined) patch.boardAccess = access;
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) { if (generalTurns !== undefined) {
reservedGeneralTurns.value = structurallyShare<unknown>( patch.reservedGeneralTurns = generalTurns.turns;
reservedGeneralTurns.value, patch.reservedGeneralRevision = generalTurns.revision;
generalTurns.turns
) as ReservedTurnView[];
reservedGeneralRevision.value = generalTurns.revision;
} }
if (records) applyRecentRecords(records); if (records) {
if (nextFrontStatus) updateFrontStatus(nextFrontStatus); const nextGlobalRecords = mergeRecentRecords(globalRecords.value, records.global);
const nextGeneralRecords = mergeRecentRecords(generalRecords.value, records.general);
const nextWorldHistory = mergeRecentRecords(worldHistory.value, records.history);
patch.globalRecords = nextGlobalRecords;
patch.generalRecords = nextGeneralRecords;
patch.worldHistory = nextWorldHistory;
}
if (nextFrontStatus) patch.frontStatus = nextFrontStatus;
applyDashboardPatch(patch);
publishDashboardPatch(patch);
} catch (err) { } catch (err) {
error.value = resolveErrorMessage(err); error.value = resolveErrorMessage(err);
} finally { } finally {
@@ -496,7 +605,10 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
return; return;
} }
try { try {
messages.value = structurallyShare(messages.value, await trpc.messages.getRecent.query({ generalId: id })); const nextMessages = await trpc.messages.getRecent.query({ generalId: id });
const patch = { messages: nextMessages } satisfies DashboardReadModelPatch;
applyDashboardPatch(patch);
publishDashboardPatch(patch);
} catch (err) { } catch (err) {
error.value = resolveErrorMessage(err); error.value = resolveErrorMessage(err);
} }
@@ -762,6 +874,61 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
realtimeToken = null; realtimeToken = null;
}; };
const isRealtimeParticipant = (): boolean =>
realtimeActive.value &&
document.visibilityState !== 'hidden' &&
realtimeEnabled.value &&
session.isReady &&
session.hasGeneral &&
generalId.value !== null;
const closeRealtimeCoordinator = () => {
const coordinator = realtimeCoordinator;
realtimeCoordinator = null;
realtimeCoordinatorScope = null;
coordinator?.stop();
closeRealtimeSource();
};
const reconcileRealtimeCoordinator = () => {
if (typeof window === 'undefined') return;
if (!isRealtimeParticipant()) {
closeRealtimeCoordinator();
return;
}
if (typeof BroadcastChannel === 'undefined') {
void connectRealtime();
return;
}
const profile = session.profile ?? 'game';
const account = session.user?.id ?? `general-${generalId.value}`;
const scope = `${encodeURIComponent(profile)}:${encodeURIComponent(account)}`;
if (realtimeCoordinator && realtimeCoordinatorScope === scope) return;
closeRealtimeCoordinator();
realtimeCoordinatorScope = scope;
realtimeCoordinator = createBroadcastTabCoordinator<DashboardTabMessage>(`sammo-main-dashboard:${scope}`, {
onLeadershipChange: (leader) => {
if (leader) {
void connectRealtime();
} else {
closeRealtimeSource();
if (realtimeEnabled.value) realtimeStatus.value = 'idle';
}
},
onPayload: (message) => {
if (!isRealtimeParticipant()) return;
if (message.kind === 'patch') {
applyDashboardPatch(message.patch);
return;
}
realtimeStatus.value = message.status;
},
});
realtimeCoordinator.start();
};
const ensureAccessToken = async (): Promise<string | null> => { const ensureAccessToken = async (): Promise<string | null> => {
if (!session.gameToken) { if (!session.gameToken) {
return null; return null;
@@ -785,7 +952,8 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
document.visibilityState === 'hidden' || document.visibilityState === 'hidden' ||
!realtimeEnabled.value || !realtimeEnabled.value ||
!session.isReady || !session.isReady ||
!session.hasGeneral !session.hasGeneral ||
(realtimeCoordinator !== null && !realtimeCoordinator.isLeader())
) { ) {
return; return;
} }
@@ -794,6 +962,9 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
realtimeStatus.value = 'idle'; realtimeStatus.value = 'idle';
return; return;
} }
if (!isRealtimeParticipant() || (realtimeCoordinator !== null && !realtimeCoordinator.isLeader())) {
return;
}
if (realtimeSource && realtimeToken === token) { if (realtimeSource && realtimeToken === token) {
return; return;
} }
@@ -806,11 +977,14 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
source.addEventListener('open', () => { source.addEventListener('open', () => {
realtimeStatus.value = 'connected'; realtimeStatus.value = 'connected';
realtimeCoordinator?.postFromLeader({ kind: 'status', status: 'connected' });
}); });
source.addEventListener('error', () => { source.addEventListener('error', () => {
realtimeStatus.value = realtimeEnabled.value ? 'idle' : 'paused'; realtimeStatus.value = realtimeEnabled.value ? 'idle' : 'paused';
realtimeCoordinator?.postFromLeader({ kind: 'status', status: 'idle' });
}); });
source.addEventListener('turnCompleted', (event) => { source.addEventListener('turnCompleted', (event) => {
if (realtimeCoordinator !== null && !realtimeCoordinator.isLeader()) return;
const payload = parseRealtimePayload(event); const payload = parseRealtimePayload(event);
if (!payload || payload.type !== 'turnCompleted') { if (!payload || payload.type !== 'turnCompleted') {
return; return;
@@ -823,6 +997,7 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
readModelRefreshQueue.request(payload.changes); readModelRefreshQueue.request(payload.changes);
}); });
source.addEventListener('readModelChanged', (event) => { source.addEventListener('readModelChanged', (event) => {
if (realtimeCoordinator !== null && !realtimeCoordinator.isLeader()) return;
const payload = parseRealtimePayload(event); const payload = parseRealtimePayload(event);
if (!payload || payload.type !== 'readModelChanged') { if (!payload || payload.type !== 'readModelChanged') {
return; return;
@@ -830,6 +1005,7 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
readModelRefreshQueue.request(payload.changes); readModelRefreshQueue.request(payload.changes);
}); });
source.addEventListener('messageCreated', (event) => { source.addEventListener('messageCreated', (event) => {
if (realtimeCoordinator !== null && !realtimeCoordinator.isLeader()) return;
const payload = parseRealtimePayload(event); const payload = parseRealtimePayload(event);
if (!payload || payload.type !== 'messageCreated') { if (!payload || payload.type !== 'messageCreated') {
return; return;
@@ -841,6 +1017,7 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
source.addEventListener('ping', () => { source.addEventListener('ping', () => {
if (realtimeEnabled.value) { if (realtimeEnabled.value) {
realtimeStatus.value = 'connected'; realtimeStatus.value = 'connected';
realtimeCoordinator?.postFromLeader({ kind: 'status', status: 'connected' });
} }
}); });
}; };
@@ -850,13 +1027,12 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
if (document.visibilityState === 'hidden') { if (document.visibilityState === 'hidden') {
realtimeRefreshQueue.cancelPending(); realtimeRefreshQueue.cancelPending();
readModelRefreshQueue.cancelPending(); readModelRefreshQueue.cancelPending();
closeRealtimeSource(); closeRealtimeCoordinator();
realtimeStatus.value = 'idle'; realtimeStatus.value = 'idle';
return; return;
} }
realtimeRefreshQueue.beginCooldown(); realtimeRefreshQueue.beginCooldown();
void connectRealtime(); void refreshQueue.request().finally(() => reconcileRealtimeCoordinator());
realtimeRefreshQueue.request();
}; };
const startRealtime = () => { const startRealtime = () => {
@@ -867,13 +1043,14 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
document.addEventListener('visibilitychange', handleVisibilityChange); document.addEventListener('visibilitychange', handleVisibilityChange);
visibilityListenerInstalled = true; visibilityListenerInstalled = true;
} }
reconcileRealtimeCoordinator();
}; };
const stopRealtime = () => { const stopRealtime = () => {
realtimeActive.value = false; realtimeActive.value = false;
realtimeRefreshQueue.cancelPending(); realtimeRefreshQueue.cancelPending();
readModelRefreshQueue.cancelPending(); readModelRefreshQueue.cancelPending();
closeRealtimeSource(); closeRealtimeCoordinator();
if (visibilityListenerInstalled) { if (visibilityListenerInstalled) {
document.removeEventListener('visibilitychange', handleVisibilityChange); document.removeEventListener('visibilitychange', handleVisibilityChange);
visibilityListenerInstalled = false; visibilityListenerInstalled = false;
@@ -882,24 +1059,22 @@ export const useMainDashboardStore = defineStore('mainDashboard', () => {
}; };
watch( watch(
() => [realtimeActive.value, realtimeEnabled.value, session.isReady, session.hasGeneral, session.gameToken], () => [
realtimeActive.value,
realtimeEnabled.value,
session.isReady,
session.hasGeneral,
session.gameToken,
session.profile,
session.user?.id,
generalId.value,
],
([active, enabled, ready, hasGeneral]) => { ([active, enabled, ready, hasGeneral]) => {
if (!active) { realtimeStatus.value = !enabled ? 'paused' : realtimeStatus.value;
closeRealtimeSource(); if (!active || !ready || !hasGeneral) {
realtimeStatus.value = enabled ? 'idle' : 'paused'; realtimeStatus.value = enabled ? 'idle' : 'paused';
return;
} }
if (!enabled) { reconcileRealtimeCoordinator();
closeRealtimeSource();
realtimeStatus.value = 'paused';
return;
}
if (!ready || !hasGeneral) {
closeRealtimeSource();
realtimeStatus.value = 'idle';
return;
}
void connectRealtime();
} }
); );
@@ -0,0 +1,166 @@
export interface BroadcastChannelLike<T> {
onmessage: ((event: MessageEvent<T>) => void) | null;
postMessage(message: T): void;
close(): void;
}
type CoordinatorWireMessage<T> =
| { kind: 'presence'; tabId: string; sentAt: number }
| { kind: 'leave'; tabId: string; sentAt: number }
| { kind: 'payload'; tabId: string; sentAt: number; payload: T };
export interface BroadcastTabCoordinator<T> {
start(): void;
stop(): void;
isLeader(): boolean;
postFromLeader(payload: T): boolean;
}
interface BroadcastTabCoordinatorOptions<T> {
onLeadershipChange: (leader: boolean) => void;
onPayload: (payload: T) => void;
createChannel?: (name: string) => BroadcastChannelLike<CoordinatorWireMessage<T>>;
createTabId?: () => string;
now?: () => number;
setTimer?: (callback: () => void, delayMs: number) => ReturnType<typeof setTimeout>;
clearTimer?: (handle: ReturnType<typeof setTimeout>) => void;
settleMs?: number;
heartbeatMs?: number;
peerExpiryMs?: number;
}
const defaultCreateTabId = (): string => {
if (typeof crypto !== 'undefined' && typeof crypto.randomUUID === 'function') {
return crypto.randomUUID();
}
return `${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}`;
};
/**
* Elect one active tab per channel without persisting credentials or state.
* Every participant advertises a short-lived presence lease; the lowest tab id
* owns realtime I/O and broadcasts the fetched result to the other tabs.
*/
export const createBroadcastTabCoordinator = <T>(
channelName: string,
options: BroadcastTabCoordinatorOptions<T>
): BroadcastTabCoordinator<T> => {
const createChannel =
options.createChannel ??
((name: string) => new BroadcastChannel(name) as BroadcastChannelLike<CoordinatorWireMessage<T>>);
const createTabId = options.createTabId ?? defaultCreateTabId;
const now = options.now ?? Date.now;
const setTimer = options.setTimer ?? ((callback, delayMs) => setTimeout(callback, delayMs));
const clearTimer = options.clearTimer ?? ((handle) => clearTimeout(handle));
const settleMs = Math.max(0, options.settleMs ?? 100);
const heartbeatMs = Math.max(100, options.heartbeatMs ?? 2_000);
const peerExpiryMs = Math.max(heartbeatMs * 2, options.peerExpiryMs ?? 5_000);
const tabId = createTabId();
let channel: BroadcastChannelLike<CoordinatorWireMessage<T>> | null = null;
let settleTimer: ReturnType<typeof setTimeout> | null = null;
let heartbeatTimer: ReturnType<typeof setTimeout> | null = null;
let started = false;
let settled = false;
let leader = false;
const peers = new Map<string, number>();
const send = (message: CoordinatorWireMessage<T>): boolean => {
if (!channel) return false;
try {
channel.postMessage(message);
return true;
} catch {
// Realtime fan-out is best effort, like the SSE/Redis boundary.
return false;
}
};
const announce = () => {
const sentAt = now();
peers.set(tabId, sentAt);
send({ kind: 'presence', tabId, sentAt });
};
const setLeader = (next: boolean) => {
if (leader === next) return;
leader = next;
options.onLeadershipChange(next);
};
const elect = () => {
if (!started || !settled) return;
const cutoff = now() - peerExpiryMs;
for (const [peerId, seenAt] of peers) {
if (peerId !== tabId && seenAt < cutoff) peers.delete(peerId);
}
peers.set(tabId, now());
const elected = [...peers.keys()].sort()[0];
setLeader(elected === tabId);
};
const scheduleHeartbeat = () => {
heartbeatTimer = setTimer(() => {
heartbeatTimer = null;
if (!started) return;
announce();
elect();
scheduleHeartbeat();
}, heartbeatMs);
};
return {
start: () => {
if (started) return;
started = true;
channel = createChannel(channelName);
channel.onmessage = (event) => {
const message = event.data;
if (!message || message.tabId === tabId) return;
if (message.kind === 'presence') {
const isNewPeer = !peers.has(message.tabId);
peers.set(message.tabId, now());
if (isNewPeer) announce();
elect();
return;
}
if (message.kind === 'leave') {
peers.delete(message.tabId);
elect();
return;
}
if (message.kind === 'payload' && !leader) {
peers.set(message.tabId, now());
options.onPayload(message.payload);
elect();
}
};
announce();
settleTimer = setTimer(() => {
settleTimer = null;
settled = true;
elect();
}, settleMs);
scheduleHeartbeat();
},
stop: () => {
if (!started) return;
send({ kind: 'leave', tabId, sentAt: now() });
started = false;
settled = false;
peers.clear();
if (settleTimer !== null) clearTimer(settleTimer);
if (heartbeatTimer !== null) clearTimer(heartbeatTimer);
settleTimer = null;
heartbeatTimer = null;
channel?.close();
channel = null;
setLeader(false);
},
isLeader: () => started && leader,
postFromLeader: (payload) => {
if (!started || !leader) return false;
return send({ kind: 'payload', tabId, sentAt: now(), payload });
},
};
};
@@ -0,0 +1,84 @@
import assert from 'node:assert/strict';
import { test } from 'node:test';
import { createBroadcastTabCoordinator } from '../src/utils/broadcastTabCoordinator.ts';
type FakeMessage = { kind: string; tabId: string; sentAt: number; payload?: string };
class FakeBroadcastBus {
readonly channels = new Set<FakeBroadcastChannel>();
open(): FakeBroadcastChannel {
const channel = new FakeBroadcastChannel(this);
this.channels.add(channel);
return channel;
}
}
class FakeBroadcastChannel {
onmessage: ((event: MessageEvent<FakeMessage>) => void) | null = null;
private readonly bus: FakeBroadcastBus;
constructor(bus: FakeBroadcastBus) {
this.bus = bus;
}
postMessage(message: FakeMessage): void {
for (const channel of this.bus.channels) {
if (channel === this) continue;
queueMicrotask(() => channel.onmessage?.({ data: message } as MessageEvent<FakeMessage>));
}
}
close(): void {
this.bus.channels.delete(this);
this.onmessage = null;
}
}
const wait = (delayMs: number) => new Promise((resolve) => setTimeout(resolve, delayMs));
void test('one active tab owns broadcasts, followers consume its payload, and leave hands leadership over', async () => {
const bus = new FakeBroadcastBus();
const firstLeadership: boolean[] = [];
const secondLeadership: boolean[] = [];
const firstPayloads: string[] = [];
const secondPayloads: string[] = [];
const common = {
createChannel: () => bus.open() as never,
settleMs: 5,
heartbeatMs: 100,
peerExpiryMs: 500,
};
const first = createBroadcastTabCoordinator<string>('same-account', {
...common,
createTabId: () => 'tab-b',
onLeadershipChange: (leader) => firstLeadership.push(leader),
onPayload: (payload) => firstPayloads.push(payload),
});
const second = createBroadcastTabCoordinator<string>('same-account', {
...common,
createTabId: () => 'tab-a',
onLeadershipChange: (leader) => secondLeadership.push(leader),
onPayload: (payload) => secondPayloads.push(payload),
});
first.start();
second.start();
await wait(20);
assert.equal(first.isLeader(), false);
assert.equal(second.isLeader(), true);
assert.equal(first.postFromLeader('ignored'), false);
assert.equal(second.postFromLeader('shared-patch'), true);
await wait(0);
assert.deepEqual(firstPayloads, ['shared-patch']);
assert.deepEqual(secondPayloads, []);
second.stop();
await wait(0);
assert.equal(first.isLeader(), true);
assert.equal(firstLeadership.at(-1), true);
assert.deepEqual(secondLeadership, [true, false]);
first.stop();
});
+7
View File
@@ -84,6 +84,13 @@ runner는 test 시작 전에 실패합니다. 지원 mode에 marker가 하나도
실행 group의 marker 정규식이 비어도 전체 파일로 선택 범위를 넓히지 않고 실행 group의 marker 정규식이 비어도 전체 파일로 선택 범위를 넓히지 않고
실패합니다. 실패합니다.
`external_fixture` mode는 격리 빈 schema로 만들 수 없는 명시적 예외입니다.
현재 `CURRENT_SEASON_FIXTURE_DATABASE_URL`은 별도 Ref 현 시즌 importer가 만든
read-only snapshot을 요구하므로 조건부 runner의 pass/skip 집계에 포함하지
않습니다. 이 mode는 registry 누락을 숨기기 위한 일반 제외 수단이 아니며,
해당 fixture 검증은 실제 imported snapshot URL을 주입해 별도로 실행하고 결과를
외부 fixture 범위로 보고해야 합니다.
관리자 시간 조정의 PostgreSQL 경계는 관리자 시간 조정의 PostgreSQL 경계는
`runtimeClockShiftPersistence.integration.test.ts`, gateway action `runtimeClockShiftPersistence.integration.test.ts`, gateway action
`PARTIAL → APPLIED` 경계는 `gatewayRuntimeAction.integration.test.ts`입니다. `PARTIAL → APPLIED` 경계는 `gatewayRuntimeAction.integration.test.ts`입니다.
@@ -1,5 +1,6 @@
# Environment variable Execution mode # Environment variable Execution mode
CREATE_GENERAL_DATABASE_URL create_general CREATE_GENERAL_DATABASE_URL create_general
CURRENT_SEASON_FIXTURE_DATABASE_URL external_fixture
GATEWAY_RUNTIME_ACTION_DATABASE_URL gateway_runtime GATEWAY_RUNTIME_ACTION_DATABASE_URL gateway_runtime
GATEWAY_OPERATION_DATABASE_URL gateway_runtime GATEWAY_OPERATION_DATABASE_URL gateway_runtime
GATEWAY_RELEASE_DATABASE_URL gateway_runtime GATEWAY_RELEASE_DATABASE_URL gateway_runtime
1 # Environment variable Execution mode
2 CREATE_GENERAL_DATABASE_URL create_general
3 CURRENT_SEASON_FIXTURE_DATABASE_URL external_fixture
4 GATEWAY_RUNTIME_ACTION_DATABASE_URL gateway_runtime
5 GATEWAY_OPERATION_DATABASE_URL gateway_runtime
6 GATEWAY_RELEASE_DATABASE_URL gateway_runtime
+12 -1
View File
@@ -41,7 +41,7 @@ node_tag=$(printf '%s' "${CI_NODE_INDEX:-local}" | tr -cd 'a-zA-Z0-9_' | tr 'A-Z
run_id=$(date -u +%m%d%H%M%S)_$$_${node_tag} run_id=$(date -u +%m%d%H%M%S)_$$_${node_tag}
export CONDITIONAL_INTEGRATION_RUN_ID=$run_id export CONDITIONAL_INTEGRATION_RUN_ID=$run_id
schema_ownership_token="sammo-conditional-integration:$run_id" schema_ownership_token="sammo-conditional-integration:$run_id"
supported_registry_modes="core create_general gateway_runtime immediate_action npc_possession reference_live_sortie reference_npc_possession select_pool" supported_registry_modes="core create_general external_fixture gateway_runtime immediate_action npc_possession reference_live_sortie reference_npc_possession select_pool"
term_grace_seconds=${CONDITIONAL_INTEGRATION_TERM_GRACE_SECONDS:-10} term_grace_seconds=${CONDITIONAL_INTEGRATION_TERM_GRACE_SECONDS:-10}
case "$term_grace_seconds" in case "$term_grace_seconds" in
''|*[!0-9]*) ''|*[!0-9]*)
@@ -88,6 +88,15 @@ done
report_dir=$(mktemp -d) report_dir=$(mktemp -d)
summary_file="$report_dir/summary.tsv" summary_file="$report_dir/summary.tsv"
validated_registry_file="$report_dir/validated-registry.tsv" validated_registry_file="$report_dir/validated-registry.tsv"
image_upload_secret_file="$report_dir/image-upload-secret"
(
umask 077
node --input-type=module -e \
'import { randomBytes } from "node:crypto"; process.stdout.write(randomBytes(32).toString("hex"));' \
>"$image_upload_secret_file"
)
export GAME_IMAGE_UPLOAD_SECRET_FILE=$image_upload_secret_file
export GATEWAY_IMAGE_UPLOAD_SECRET_FILE=$image_upload_secret_file
active_process_group= active_process_group=
cleanup_resources_started=0 cleanup_resources_started=0
@@ -452,7 +461,9 @@ validate_marker_registry
pnpm install --frozen-lockfile pnpm install --frozen-lockfile
pnpm --filter @sammo-ts/infra prisma:generate pnpm --filter @sammo-ts/infra prisma:generate
pnpm --filter @sammo-ts/common build pnpm --filter @sammo-ts/common build
pnpm --filter @sammo-ts/logic build
pnpm --filter @sammo-ts/infra build pnpm --filter @sammo-ts/infra build
pnpm --filter @sammo-ts/game-engine build
GATEWAY_MIGRATION_TEST_DATABASE_URL=$base_database_url \ GATEWAY_MIGRATION_TEST_DATABASE_URL=$base_database_url \
pnpm --filter @sammo-ts/infra verify:migration:account-icon pnpm --filter @sammo-ts/infra verify:migration:account-icon