diff --git a/app/game-api/src/context.ts b/app/game-api/src/context.ts index 99f61b54..2d3393f2 100644 --- a/app/game-api/src/context.ts +++ b/app/game-api/src/context.ts @@ -89,8 +89,6 @@ export interface GameApiContext { generalAccessTracking?: boolean; /** Validated server-issued proof for one realtime refresh burst. */ realtimeAccessGranted?: boolean; - /** Request-local identity already resolved by the realtime access gate. */ - realtimeAccessGeneralId?: number; /** Set only while an API input-event transaction owns the mutation. */ changeJournal?: ChangeJournal; /** Post-commit scheduling hint for the durable outbox dispatcher. */ diff --git a/app/game-api/src/router/dashboard/index.ts b/app/game-api/src/router/dashboard/index.ts index 299fc941..875002aa 100644 --- a/app/game-api/src/router/dashboard/index.ts +++ b/app/game-api/src/router/dashboard/index.ts @@ -2,20 +2,19 @@ import { TRPCError } from '@trpc/server'; import type { ReadModelDelta } from '@sammo-ts/common'; import { z } from 'zod'; -import { accessLimitAuthedProcedure, router } from '../../trpc.js'; +import { deferredAccessLimitAuthedProcedure, router } from '../../trpc.js'; import { canUseDashboardSourceRevision, - readDashboardSourceRevisionState, + readDashboardSourceRevisionStateForUser, type DashboardSourceRevisionState, type DashboardSourceSlice, } from '../../services/dashboardSourceRevision.js'; import { createReadModelDelta } from '../../services/readModelDeltaCache.js'; import { DASHBOARD_PROJECTION_ACCESS_WEIGHT, - formatGeneralAccessLimitMessage, - getGeneralAccessState, recordGeneralAccessWeight, } from '../../services/generalAccess.js'; +import { enqueueDeferredGeneralAccess } from '../../services/deferredGeneralAccess.js'; import { getBoardAccess } from '../board/index.js'; import { getGeneralContext } from '../general/index.js'; import { getTurnCommandTable } from '../turns/index.js'; @@ -102,7 +101,7 @@ const createDashboardSliceDelta = async (options: { }; export const dashboardRouter = router({ - getContextBundleDelta: accessLimitAuthedProcedure.input(zContextBundleInput).query(async ({ ctx, input }) => { + getContextBundleDelta: deferredAccessLimitAuthedProcedure.input(zContextBundleInput).query(async ({ ctx, input }) => { const authUser = ctx.auth?.user; const viewerId = authUser?.id; if (!authUser || !viewerId) { @@ -110,10 +109,13 @@ export const dashboardRouter = router({ } const includesProjection = Object.values(input.include).some(Boolean); - let generalId: number | null = null; + const sourceState = includesProjection + ? await readDashboardSourceRevisionStateForUser(ctx.db, viewerId, authUser) + : null; + let generalId: number | null = sourceState?.identity.generalId ?? null; if (includesProjection) { generalId = - ctx.realtimeAccessGeneralId ?? + generalId ?? ( await ctx.db.general.findFirst({ where: { userId: viewerId }, @@ -123,7 +125,6 @@ export const dashboardRouter = router({ )?.id ?? null; } - let sourceState = generalId ? await readDashboardSourceRevisionState(ctx.db, generalId, authUser) : null; const buildSliceRequests = (): DashboardSliceRequest[] => [ { included: input.include.context, @@ -152,17 +153,21 @@ export const dashboardRouter = router({ ]; const sliceRequests = buildSliceRequests(); const rebuildsPostgresProjection = sliceRequests.some(requiresDashboardProjection); - if (rebuildsPostgresProjection && ctx.generalAccessTracking === true && ctx.realtimeAccessGranted !== true) { - const recorded = await recordGeneralAccessWeight(ctx, DASHBOARD_PROJECTION_ACCESS_WEIGHT); - const accessState = await getGeneralAccessState(ctx); - if (accessState?.level === 2) { - throw new TRPCError({ - code: 'TOO_MANY_REQUESTS', - message: formatGeneralAccessLimitMessage(accessState), - }); - } - if (recorded && generalId !== null) { - sourceState = await readDashboardSourceRevisionState(ctx.db, generalId, authUser); + if (ctx.generalAccessTracking === true) { + if (rebuildsPostgresProjection && ctx.realtimeAccessGranted !== true && generalId !== null) { + try { + await enqueueDeferredGeneralAccess( + ctx.redis, + ctx.profile.name, + ctx.auth, + generalId, + DASHBOARD_PROJECTION_ACCESS_WEIGHT + ); + } catch { + // Redis is the low-cost path. Preserve the access record when it is + // unavailable, accepting the older synchronous SQL only on failure. + await recordGeneralAccessWeight(ctx, DASHBOARD_PROJECTION_ACCESS_WEIGHT); + } } } diff --git a/app/game-api/src/router/general/index.ts b/app/game-api/src/router/general/index.ts index 8d3cc216..6dfc5bac 100644 --- a/app/game-api/src/router/general/index.ts +++ b/app/game-api/src/router/general/index.ts @@ -8,8 +8,8 @@ import type { GameApiContext } from '../../context.js'; import { accessEngineAuthedProcedure, accessEngineAuthedInputProcedure, - accessLimitAuthedProcedure, authedProcedure, + deferredAccessLimitAuthedProcedure, engineAuthedProcedure, router, } from '../../trpc.js'; @@ -739,7 +739,7 @@ export const generalRouter = router({ })), }; }), - getRecentRecords: accessLimitAuthedProcedure + getRecentRecords: deferredAccessLimitAuthedProcedure .input( z.object({ lastGeneralRecordId: z.number().int().nonnegative().default(0), @@ -792,7 +792,7 @@ export const generalRouter = router({ // 메인 화면은 SSE invalidation, 탭 복귀와 직접 갱신이 같은 read model을 // 호출한다. 클라이언트가 주장하는 갱신 원인을 신뢰해 구분하지 않고 이 // projection 전체를 무가점으로 두되, 이미 제한된 사용자의 gate는 유지한다. - getFrontStatus: accessLimitAuthedProcedure.query(async ({ ctx }) => { + getFrontStatus: deferredAccessLimitAuthedProcedure.query(async ({ ctx }) => { const me = await getMyGeneral(ctx); const worldState = await ctx.db.worldState.findFirst({ orderBy: { id: 'asc' }, diff --git a/app/game-api/src/server.ts b/app/game-api/src/server.ts index c586ff07..6854c5a1 100644 --- a/app/game-api/src/server.ts +++ b/app/game-api/src/server.ts @@ -41,6 +41,7 @@ import { AccountIconResetReconciler } from './services/accountIconResetReconcile import { createBestEffortResourceCloser } from './services/bestEffortResourceCloser.js'; import { RemoteContentImageStore } from './services/remoteContentImageStore.js'; import { ReadModelOutboxWorker } from './realtime/outboxWorker.js'; +import { DeferredGeneralAccessWorker } from './services/deferredGeneralAccess.js'; const extractBearerToken = (value: string | string[] | undefined): string | null => { if (!value) { @@ -154,9 +155,22 @@ export const createGameApiServer = async () => { const readModelOutboxWorker = new ReadModelOutboxWorker(postgres.prisma, redis.client, config.profileName, { onError: (error) => app.log.error({ err: error }, 'read-model outbox dispatch failed'), }); + const deferredGeneralAccessWorker = new DeferredGeneralAccessWorker( + postgres.prisma, + redis.client, + config.profileName, + profileStatusSource, + { + onError: (error) => app.log.error({ err: error }, 'deferred general access flush failed'), + } + ); let flushSubscriberStarted = false; let realtimeHubStarted = false; const closeResources = createBestEffortResourceCloser([ + { + name: 'deferred-general-access-worker', + run: () => deferredGeneralAccessWorker.stop(), + }, { name: 'read-model-outbox-worker', run: () => readModelOutboxWorker.stop(), @@ -374,6 +388,7 @@ export const createGameApiServer = async () => { await flushSubscriber.start(); flushSubscriberStarted = true; readModelOutboxWorker.start(); + deferredGeneralAccessWorker.start(); accountIconResetReconciler.start(); } catch (error) { await closeResources(); diff --git a/app/game-api/src/services/dashboardSourceRevision.ts b/app/game-api/src/services/dashboardSourceRevision.ts index 0ca57571..a74e58dc 100644 --- a/app/game-api/src/services/dashboardSourceRevision.ts +++ b/app/game-api/src/services/dashboardSourceRevision.ts @@ -12,7 +12,7 @@ import type { DatabaseClient } from '../context.js'; export const DASHBOARD_SOURCE_REVISION_COVERAGE_VERSION = 1; const SOURCE_REVISION_LENGTH = 22; -const SOURCE_REVISION_CODE_VERSION = 'dashboard-private-slices-v2'; +const SOURCE_REVISION_CODE_VERSION = 'dashboard-private-slices-v3'; export type DashboardSourceSlice = 'context' | 'commandTable' | 'boardAccess'; @@ -36,7 +36,6 @@ interface DashboardSourceRevisionRow { cityRevision: bigint; nationRevision: bigint; worldRevision: bigint; - accessRevision: bigint; } type RevisionTuple = readonly [domain: string, entityId: number, revision: string]; @@ -46,7 +45,6 @@ type DashboardRevisionVector = { city: string; nation: string; world: string; - access: string; }; export interface DashboardAuthSource { @@ -124,11 +122,10 @@ const buildSourceRevisions = ( const city = ['city.content', identity.cityId, identity.cityId > 0 ? revisions.city : '0'] as const; const nation = ['nation.content', identity.nationId, identity.nationId > 0 ? revisions.nation : '0'] as const; const world = ['world.content', 0, revisions.world] as const; - const access = ['access.general', identity.generalId, revisions.access] as const; const auth = ['auth.context', 0, authSourceRevision] as const; return { - context: digestSourceRevision('context', [global, general, city, nation, world, access, auth]), + context: digestSourceRevision('context', [global, general, city, nation, world, auth]), commandTable: digestSourceRevision('commandTable', [global, general, city, nation, world]), boardAccess: digestSourceRevision('boardAccess', [general, nation]), }; @@ -139,15 +136,11 @@ const buildSourceRevisions = ( * dependency heads in one indexed statement. Missing revision rows are revision 0. * A missing actor/meta row, query failure, or malformed result disables the optimization. */ -export const readDashboardSourceRevisionState = async ( +const readDashboardSourceRevisionStateByActor = async ( db: Pick, - generalId: number, + actorCondition: GamePrisma.Sql, authSource?: DashboardAuthSource ): Promise => { - if (!Number.isSafeInteger(generalId) || generalId <= 0) { - return null; - } - let rows: DashboardSourceRevisionRow[]; try { rows = await db.$queryRaw(GamePrisma.sql` @@ -160,8 +153,7 @@ export const readDashboardSourceRevisionState = async ( COALESCE(general_revision."revision", 0) AS "generalRevision", COALESCE(city_revision."revision", 0) AS "cityRevision", COALESCE(nation_revision."revision", 0) AS "nationRevision", - COALESCE(world_revision."revision", 0) AS "worldRevision", - COALESCE(access_revision."revision", 0) AS "accessRevision" + COALESCE(world_revision."revision", 0) AS "worldRevision" FROM "general" AS actor CROSS JOIN "read_model_revision_meta" AS meta LEFT JOIN "read_model_revision" AS global_revision @@ -179,10 +171,7 @@ export const readDashboardSourceRevisionState = async ( LEFT JOIN "read_model_revision" AS world_revision ON world_revision."domain" = 'world.content' AND world_revision."entity_id" = 0 - LEFT JOIN "read_model_revision" AS access_revision - ON access_revision."domain" = 'access.general' - AND access_revision."entity_id" = actor."id" - WHERE actor."id" = ${generalId} + WHERE ${actorCondition} AND meta."id" = 1 `); } catch { @@ -206,10 +195,9 @@ export const readDashboardSourceRevisionState = async ( city: parseNonNegativeRevision(row.cityRevision), nation: parseNonNegativeRevision(row.nationRevision), world: parseNonNegativeRevision(row.worldRevision), - access: parseNonNegativeRevision(row.accessRevision), }; if ( - identity.generalId !== generalId || + identity.generalId === null || identity.cityId === null || identity.nationId === null || coverageVersion === null || @@ -219,7 +207,7 @@ export const readDashboardSourceRevisionState = async ( } const validIdentity = { - generalId, + generalId: identity.generalId, cityId: identity.cityId, nationId: identity.nationId, }; @@ -234,6 +222,28 @@ export const readDashboardSourceRevisionState = async ( }; }; +export const readDashboardSourceRevisionState = async ( + db: Pick, + generalId: number, + authSource?: DashboardAuthSource +): Promise => { + if (!Number.isSafeInteger(generalId) || generalId <= 0) { + return null; + } + return readDashboardSourceRevisionStateByActor(db, GamePrisma.sql`actor."id" = ${generalId}`, authSource); +}; + +export const readDashboardSourceRevisionStateForUser = async ( + db: Pick, + userId: string, + authSource?: DashboardAuthSource +): Promise => { + if (!userId) { + return null; + } + return readDashboardSourceRevisionStateByActor(db, GamePrisma.sql`actor."user_id" = ${userId}`, authSource); +}; + export const canUseDashboardSourceRevision = (options: { state: DashboardSourceRevisionState | null; slice: DashboardSourceSlice; diff --git a/app/game-api/src/services/deferredGeneralAccess.ts b/app/game-api/src/services/deferredGeneralAccess.ts new file mode 100644 index 00000000..4e6b9389 --- /dev/null +++ b/app/game-api/src/services/deferredGeneralAccess.ts @@ -0,0 +1,535 @@ +import { createHash, randomUUID } from 'node:crypto'; + +import { asRecord, resolveAccessLimitLevel, resolveAccessRefreshLimit } from '@sammo-ts/common'; +import { GamePrisma } from '@sammo-ts/infra'; +import type { GameSessionTokenPayload } from '@sammo-ts/common/auth/gameToken'; + +import type { DatabaseClient } from '../context.js'; +import type { ProfileStatusSource } from '../auth/profileStatusSource.js'; +import { resolveAccessWindows } from './generalAccess.js'; + +export const DEFERRED_GENERAL_ACCESS_FLUSH_INTERVAL_MS = 5_000; +const ACTIVE_BATCH_TTL_MS = 24 * 60 * 60 * 1_000; +const COMPLETED_BATCH_RETENTION_DAYS = 7; + +const adminRoles = new Set(['superuser', 'admin', 'admin.superuser']); + +interface DeferredAccessRedis { + eval(script: string, options: { keys: string[]; arguments: string[] }): Promise; + scanIterator(options: { MATCH: string; COUNT: number }): AsyncGenerator; + hGetAll(key: string): Promise>; + get(key: string): Promise; + set(key: string, value: string, options: { PX: number }): Promise; + del(key: string): Promise; +} + +export type DeferredGeneralAccessEntry = { + generalId: number; + userId: string; + weight: number; + lastRefresh: Date; +}; + +export type DeferredGeneralAccessLimit = { + nextAccessAt: Date; +}; + +type DeferredGeneralAccessFlushRow = { + generalId: number; + userId: string; + refreshScore: number; + nextAccessAt: Date; +}; + +const ENQUEUE_SCRIPT = ` +local weight_field = 'weight:' .. ARGV[1] +local user_field = 'user:' .. ARGV[1] +local time_field = 'time:' .. ARGV[1] +local next_weight = redis.call('HINCRBY', KEYS[1], weight_field, ARGV[3]) +redis.call('HSET', KEYS[1], user_field, ARGV[2]) +local current_time = redis.call('HGET', KEYS[1], time_field) +if not current_time or tonumber(ARGV[4]) > tonumber(current_time) then + redis.call('HSET', KEYS[1], time_field, ARGV[4]) +end +redis.call('PEXPIRE', KEYS[1], ARGV[5]) +return next_weight +`; + +const ROTATE_SCRIPT = ` +if redis.call('EXISTS', KEYS[1]) == 0 then + return 0 +end +if redis.call('EXISTS', KEYS[2]) == 1 then + return -1 +end +redis.call('RENAME', KEYS[1], KEYS[2]) +return 1 +`; + +const activeKey = (profileName: string): string => `sammo:game:general-access:pending:${profileName}`; +const batchPrefix = (profileName: string): string => `sammo:game:general-access:batch:${profileName}:`; +const batchKey = (profileName: string, batchId: string): string => `${batchPrefix(profileName)}${batchId}`; +const limitKey = (profileName: string, userId: string): string => + `sammo:game:general-access:limited:${profileName}:${createHash('sha256').update(userId).digest('base64url')}`; + +const isEligibleUser = (auth: GameSessionTokenPayload | null): auth is GameSessionTokenPayload => + Boolean(auth && !auth.user.roles.some((role) => adminRoles.has(role))); + +export const enqueueDeferredGeneralAccess = async ( + redis: Pick, + profileName: string, + auth: GameSessionTokenPayload | null, + generalId: number, + weight: number, + now = new Date() +): Promise => { + if (!Number.isSafeInteger(generalId) || generalId <= 0) { + return false; + } + if (!Number.isInteger(weight) || weight < 0) { + throw new RangeError('Deferred general access weight must be a non-negative integer.'); + } + if (!isEligibleUser(auth)) { + return false; + } + await redis.eval(ENQUEUE_SCRIPT, { + keys: [activeKey(profileName)], + arguments: [String(generalId), auth.user.id, String(weight), String(now.getTime()), String(ACTIVE_BATCH_TTL_MS)], + }); + return true; +}; + +export const getDeferredGeneralAccessLimit = async ( + redis: Pick, + profileName: string, + auth: GameSessionTokenPayload | null, + now = new Date() +): Promise => { + if (!isEligibleUser(auth)) { + return null; + } + const raw = await redis.get(limitKey(profileName, auth.user.id)); + if (!raw) { + return null; + } + try { + const parsed = JSON.parse(raw) as { nextAccessAt?: unknown }; + if (typeof parsed.nextAccessAt !== 'string') { + return null; + } + const nextAccessAt = new Date(parsed.nextAccessAt); + if (Number.isNaN(nextAccessAt.getTime()) || nextAccessAt.getTime() <= now.getTime()) { + return null; + } + return { nextAccessAt }; + } catch { + return null; + } +}; + +const parseBatchHash = (values: Record): DeferredGeneralAccessEntry[] => { + const result: DeferredGeneralAccessEntry[] = []; + for (const [field, rawWeight] of Object.entries(values)) { + if (!field.startsWith('weight:')) continue; + const idText = field.slice('weight:'.length); + const generalId = Number(idText); + const weight = Number(rawWeight); + const userId = values[`user:${idText}`]; + const timestamp = Number(values[`time:${idText}`]); + if ( + !Number.isSafeInteger(generalId) || + generalId <= 0 || + !Number.isSafeInteger(weight) || + weight < 0 || + !userId || + !Number.isSafeInteger(timestamp) + ) { + continue; + } + const lastRefresh = new Date(timestamp); + if (!Number.isNaN(lastRefresh.getTime())) { + result.push({ generalId, userId, weight, lastRefresh }); + } + } + return result.sort((left, right) => left.generalId - right.generalId); +}; + +const markBatchProcessed = async (db: Pick, batchId: string): Promise => { + await db.$executeRaw(GamePrisma.sql` + INSERT INTO "general_access_batch" ("id") + VALUES (${batchId}) + ON CONFLICT ("id") DO NOTHING + `); +}; + +export const flushDeferredGeneralAccessBatch = async ( + db: Pick, + batchId: string, + entries: readonly DeferredGeneralAccessEntry[] +): Promise<{ states: DeferredGeneralAccessFlushRow[]; refreshLimit: number }> => { + if (entries.length === 0) { + await markBatchProcessed(db, batchId); + return { states: [], refreshLimit: 0 }; + } + const worldState = await db.worldState.findFirst({ + orderBy: { id: 'asc' }, + select: { + id: true, + currentYear: true, + currentMonth: true, + tickSeconds: true, + meta: true, + }, + }); + if (!worldState) { + throw new Error('Deferred general access flush requires a world state.'); + } + + const meta = asRecord(worldState.meta); + const isUnited = Number(meta.isUnited ?? meta.isunited ?? 0); + const openTime = typeof meta.opentime === 'string' ? new Date(meta.opentime) : null; + const newestRefresh = entries.reduce( + (latest, entry) => (entry.lastRefresh.getTime() > latest.getTime() ? entry.lastRefresh : latest), + entries[0]!.lastRefresh + ); + if (isUnited === 2 || (openTime && !Number.isNaN(openTime.getTime()) && openTime > newestRefresh)) { + await markBatchProcessed(db, batchId); + return { + states: [], + refreshLimit: resolveAccessRefreshLimit(worldState.tickSeconds, meta.refreshLimit), + }; + } + + const { periodStartedAt } = resolveAccessWindows(newestRefresh, worldState.tickSeconds, meta); + const payload = JSON.stringify( + entries.map(({ generalId, userId, weight, lastRefresh }) => ({ + generalId, + userId, + weight, + lastRefresh: lastRefresh.toISOString(), + })) + ); + const periodKey = worldState.currentYear * 12 + worldState.currentMonth - 1; + const states = await db.$queryRaw(GamePrisma.sql` + WITH inserted_batch AS ( + INSERT INTO "general_access_batch" ("id") + VALUES (${batchId}) + ON CONFLICT ("id") DO NOTHING + RETURNING "id" + ), + input AS ( + SELECT + value."generalId" AS "general_id", + value."userId" AS "user_id", + value."weight", + value."lastRefresh" AS "last_refresh" + FROM jsonb_to_recordset(CAST(${payload} AS jsonb)) AS value( + "generalId" INTEGER, + "userId" TEXT, + "weight" INTEGER, + "lastRefresh" TIMESTAMPTZ + ) + ), + resolved AS ( + SELECT + input."general_id", + input."user_id", + input."weight", + input."last_refresh", + actor."turn_time" + FROM input + JOIN "general" AS actor + ON actor."id" = input."general_id" + AND actor."user_id" = input."user_id" + WHERE EXISTS (SELECT 1 FROM inserted_batch) + AND input."weight" >= 0 + ), + latest_period AS ( + SELECT MAX("year" * 12 + "month" - 1)::INTEGER AS "period_key" + FROM "traffic_period" + WHERE "world_state_id" = ${worldState.id} + ), + missing_periods AS ( + INSERT INTO "traffic_period" ( + "world_state_id", "year", "month", "started_at", "last_refresh", "refresh", "online" + ) + SELECT + ${worldState.id}, + (missing_key / 12)::INTEGER, + (missing_key % 12 + 1)::INTEGER, + CAST(${periodStartedAt} AS TIMESTAMP) - ( + (${periodKey} - missing_key) * ${worldState.tickSeconds} + ) * INTERVAL '1 second', + CAST(${periodStartedAt} AS TIMESTAMP) - ( + (${periodKey} - missing_key - 1) * ${worldState.tickSeconds} + ) * INTERVAL '1 second', + 0, + 0 + FROM latest_period + CROSS JOIN LATERAL generate_series(latest_period."period_key" + 1, ${periodKey} - 1) AS missing_key + WHERE latest_period."period_key" IS NOT NULL + AND EXISTS (SELECT 1 FROM resolved) + ON CONFLICT ("world_state_id", "year", "month") DO NOTHING + RETURNING "id" + ), + period_upsert AS ( + INSERT INTO "traffic_period" ( + "world_state_id", "year", "month", "started_at", "last_refresh", "refresh", "online" + ) + SELECT + ${worldState.id}, + ${worldState.currentYear}, + ${worldState.currentMonth}, + ${periodStartedAt}, + MAX(resolved."last_refresh"), + SUM(resolved."weight")::INTEGER, + COUNT(*)::INTEGER + FROM resolved + HAVING COUNT(*) > 0 + ON CONFLICT ("world_state_id", "year", "month") DO UPDATE SET + "started_at" = LEAST("traffic_period"."started_at", EXCLUDED."started_at"), + "last_refresh" = GREATEST("traffic_period"."last_refresh", EXCLUDED."last_refresh"), + "refresh" = "traffic_period"."refresh" + EXCLUDED."refresh", + "online" = "traffic_period"."online" + ( + SELECT COUNT(*)::INTEGER + FROM resolved + WHERE NOT EXISTS ( + SELECT 1 + FROM "traffic_period_general" AS existing_member + WHERE existing_member."period_id" = "traffic_period"."id" + AND existing_member."general_id" = resolved."general_id" + ) + ) + RETURNING "id" + ), + inserted_generals AS ( + INSERT INTO "traffic_period_general" ( + "period_id", "general_id", "user_id", "refresh", "last_refresh" + ) + SELECT + period_upsert."id", + resolved."general_id", + resolved."user_id", + resolved."weight", + resolved."last_refresh" + FROM resolved + CROSS JOIN period_upsert + ON CONFLICT ("period_id", "general_id") DO NOTHING + RETURNING "period_id", "general_id" + ), + updated_generals AS ( + UPDATE "traffic_period_general" AS existing + SET + "user_id" = resolved."user_id", + "refresh" = existing."refresh" + resolved."weight", + "last_refresh" = GREATEST(existing."last_refresh", resolved."last_refresh") + FROM resolved + CROSS JOIN period_upsert + WHERE existing."period_id" = period_upsert."id" + AND existing."general_id" = resolved."general_id" + AND NOT EXISTS ( + SELECT 1 FROM inserted_generals + WHERE inserted_generals."period_id" = existing."period_id" + AND inserted_generals."general_id" = existing."general_id" + ) + RETURNING existing."general_id" + ), + access_updates AS ( + INSERT INTO "general_access_log" ( + "general_id", "user_id", "last_refresh", "refresh", "refresh_total", + "refresh_score", "refresh_score_total" + ) + SELECT + resolved."general_id", + resolved."user_id", + resolved."last_refresh", + resolved."weight", + resolved."weight", + CASE + WHEN resolved."last_refresh" >= resolved."turn_time" - ( + ${worldState.tickSeconds} * INTERVAL '1 second' + ) THEN resolved."weight" + ELSE 0 + END, + resolved."weight" + FROM resolved + ON CONFLICT ("general_id") DO UPDATE SET + "user_id" = EXCLUDED."user_id", + "last_refresh" = GREATEST("general_access_log"."last_refresh", EXCLUDED."last_refresh"), + "refresh" = CASE + WHEN "general_access_log"."last_refresh" IS NULL + OR "general_access_log"."last_refresh" < ${periodStartedAt} + THEN EXCLUDED."refresh" + ELSE "general_access_log"."refresh" + EXCLUDED."refresh" + END, + "refresh_total" = "general_access_log"."refresh_total" + EXCLUDED."refresh_total", + "refresh_score" = CASE + WHEN "general_access_log"."last_refresh" IS NULL + OR "general_access_log"."last_refresh" < ( + SELECT resolved."turn_time" - (${worldState.tickSeconds} * INTERVAL '1 second') + FROM resolved + WHERE resolved."general_id" = EXCLUDED."general_id" + ) + THEN EXCLUDED."refresh_score" + ELSE "general_access_log"."refresh_score" + EXCLUDED."refresh_score" + END, + "refresh_score_total" = + "general_access_log"."refresh_score_total" + EXCLUDED."refresh_score_total" + RETURNING "general_id", "user_id", "refresh_score" + ), + pruned_batches AS ( + DELETE FROM "general_access_batch" + WHERE "created_at" < CURRENT_TIMESTAMP - (${COMPLETED_BATCH_RETENTION_DAYS} * INTERVAL '1 day') + RETURNING "id" + ) + SELECT + actor."id" AS "generalId", + input."user_id" AS "userId", + COALESCE(access_updates."refresh_score", access_log."refresh_score", 0) AS "refreshScore", + actor."turn_time" AS "nextAccessAt" + FROM input + JOIN "general" AS actor + ON actor."id" = input."general_id" + AND actor."user_id" = input."user_id" + LEFT JOIN access_updates ON access_updates."general_id" = actor."id" + LEFT JOIN "general_access_log" AS access_log ON access_log."general_id" = actor."id" + ORDER BY actor."id" + `); + return { + states, + refreshLimit: resolveAccessRefreshLimit(worldState.tickSeconds, meta.refreshLimit), + }; +}; + +export interface DeferredGeneralAccessWorkerOptions { + intervalMs?: number; + now?: () => Date; + createBatchId?: () => string; + onError?: (error: unknown) => void; +} + +export class DeferredGeneralAccessWorker { + private timer: NodeJS.Timeout | null = null; + private inFlight: Promise | null = null; + private running = false; + private readonly intervalMs: number; + private readonly now: () => Date; + private readonly createBatchId: () => string; + private readonly onError: (error: unknown) => void; + + constructor( + private readonly db: Pick, + private readonly redis: DeferredAccessRedis, + private readonly profileName: string, + private readonly profileStatusSource: ProfileStatusSource, + options: DeferredGeneralAccessWorkerOptions = {} + ) { + this.intervalMs = Math.max(1, Math.floor(options.intervalMs ?? DEFERRED_GENERAL_ACCESS_FLUSH_INTERVAL_MS)); + this.now = options.now ?? (() => new Date()); + this.createBatchId = options.createBatchId ?? randomUUID; + this.onError = options.onError ?? (() => undefined); + } + + private async listBatchKeys(): Promise { + const keys = new Set(); + for await (const page of this.redis.scanIterator({ MATCH: `${batchPrefix(this.profileName)}*`, COUNT: 100 })) { + for (const key of page) keys.add(key); + } + const batchId = this.createBatchId(); + const rotatedKey = batchKey(this.profileName, batchId); + const rotated = Number( + await this.redis.eval(ROTATE_SCRIPT, { + keys: [activeKey(this.profileName), rotatedKey], + arguments: [], + }) + ); + if (rotated === 1) keys.add(rotatedKey); + return [...keys].sort(); + } + + private async applyLimitStates( + states: readonly DeferredGeneralAccessFlushRow[], + refreshLimit: number + ): Promise { + const now = this.now(); + await Promise.all( + states.map(async (state) => { + const key = limitKey(this.profileName, state.userId); + if ( + resolveAccessLimitLevel(state.refreshScore, refreshLimit) !== 2 || + state.nextAccessAt.getTime() <= now.getTime() + ) { + await this.redis.del(key); + return; + } + await this.redis.set( + key, + JSON.stringify({ nextAccessAt: state.nextAccessAt.toISOString() }), + { PX: Math.max(1, state.nextAccessAt.getTime() - now.getTime()) } + ); + }) + ); + } + + private async processBatch(key: string, runningProfile: boolean): Promise { + if (!runningProfile) { + await this.redis.del(key); + return; + } + const values = await this.redis.hGetAll(key); + const entries = parseBatchHash(values); + const id = key.slice(batchPrefix(this.profileName).length); + if (!id) { + return; + } + const result = await flushDeferredGeneralAccessBatch(this.db, id, entries); + await this.applyLimitStates(result.states, result.refreshLimit); + await this.redis.del(key); + } + + private async runOnce(): Promise { + const keys = await this.listBatchKeys(); + if (keys.length === 0) return; + let runningProfile = false; + try { + runningProfile = (await this.profileStatusSource.get(this.profileName)) === 'RUNNING'; + } catch { + // Preserve the synchronous fail-open contract: if Gateway status + // cannot be established, do not retain a batch that may penalize + // the user after the status service recovers. + } + for (const key of keys) { + await this.processBatch(key, runningProfile); + } + } + + async flushOnce(): Promise { + if (this.inFlight) { + await this.inFlight; + return; + } + this.inFlight = this.runOnce().finally(() => { + this.inFlight = null; + }); + await this.inFlight; + } + + start(): void { + if (this.running) return; + this.running = true; + this.timer = setInterval(() => { + void this.flushOnce().catch(this.onError); + }, this.intervalMs); + this.timer.unref?.(); + void this.flushOnce().catch(this.onError); + } + + async stop(): Promise { + this.running = false; + if (this.timer) { + clearInterval(this.timer); + this.timer = null; + } + await this.inFlight; + } +} diff --git a/app/game-api/src/services/generalAccess.ts b/app/game-api/src/services/generalAccess.ts index c6d8737c..60a94b1f 100644 --- a/app/game-api/src/services/generalAccess.ts +++ b/app/game-api/src/services/generalAccess.ts @@ -1,5 +1,5 @@ import { asRecord, resolveAccessLimitLevel, resolveAccessRefreshLimit, type AccessLimitLevel } from '@sammo-ts/common'; -import { GamePrisma, writeReadModelChangeJournal } from '@sammo-ts/infra'; +import { GamePrisma } from '@sammo-ts/infra'; import type { GameApiContext } from '../context.js'; @@ -371,8 +371,6 @@ export const upsertGeneralAccess = async ( general_access_log.refresh_score_total + EXCLUDED.refresh_score_total ` ); - - await writeReadModelChangeJournal(transaction, [{ domain: 'access.general', entityId: input.generalId }]); }); }; diff --git a/app/game-api/src/trpc.ts b/app/game-api/src/trpc.ts index 9a411680..476982d9 100644 --- a/app/game-api/src/trpc.ts +++ b/app/game-api/src/trpc.ts @@ -15,6 +15,7 @@ import { resolveGeneralAccessEndpointWeight, type GeneralAccessEndpoint, } from './services/generalAccess.js'; +import { getDeferredGeneralAccessLimit } from './services/deferredGeneralAccess.js'; const t = initTRPC.context().create(); @@ -126,12 +127,21 @@ const generalAccessLimitMiddleware = t.middleware(async ({ ctx, next }) => { message: formatGeneralAccessLimitMessage(state), }); } - return next({ - ctx: { - ...ctx, - ...(state ? { realtimeAccessGeneralId: state.generalId } : {}), - }, - }); + return next(); +}); + +const deferredGeneralAccessLimitMiddleware = t.middleware(async ({ ctx, next }) => { + if (ctx.generalAccessTracking !== true) { + return next(); + } + const state = await getDeferredGeneralAccessLimit(ctx.redis, ctx.profile.name, ctx.auth); + if (state) { + throw new TRPCError({ + code: 'TOO_MANY_REQUESTS', + message: formatGeneralAccessLimitMessage(state), + }); + } + return next(); }); export const router = t.router; @@ -165,6 +175,9 @@ export const readOnlyAuthedProcedure: typeof procedure = t.procedure.use(require export const accessLimitAuthedProcedure: typeof procedure = t.procedure .use(requireAuthMiddleware) .use(generalAccessLimitMiddleware); +export const deferredAccessLimitAuthedProcedure: typeof procedure = t.procedure + .use(requireAuthMiddleware) + .use(deferredGeneralAccessLimitMiddleware); // 입력이 있는 Ref handler는 request parsing을 마친 뒤 increaseRefresh()를 // 호출한다. 이 factory들은 parser를 access/input-event middleware 앞에 둔다. export const accessInputProcedure: typeof procedure.input = (input) => diff --git a/app/game-api/test/dashboardRouter.test.ts b/app/game-api/test/dashboardRouter.test.ts index 139cd430..38ea547b 100644 --- a/app/game-api/test/dashboardRouter.test.ts +++ b/app/game-api/test/dashboardRouter.test.ts @@ -73,6 +73,7 @@ const buildContext = (authenticated: boolean, generalAccessTracking = false) => generalAccessTracking, redis: { get: async (key: string) => redisValues.get(key) ?? null, + eval: async () => 1, set: async (key: string, value: string) => { redisValues.set(key, value); return 'OK'; @@ -109,7 +110,6 @@ const installSourceRevisionState = ( cityRevision: bigint; nationRevision: bigint; worldRevision: bigint; - accessRevision: bigint; }> = {} ) => { let row = { @@ -122,12 +122,10 @@ const installSourceRevisionState = ( cityRevision: 0n, nationRevision: 0n, worldRevision: 1n, - accessRevision: 1n, ...initial, }; const queryRaw = vi.fn(async () => [row]); Object.assign(context.db, { $queryRaw: queryRaw }); - context.realtimeAccessGeneralId = 7; return { queryRaw, update: (next: Partial) => { @@ -227,7 +225,7 @@ describe('dashboardRouter.getContextBundleDelta', () => { ).rejects.toMatchObject({ code: 'UNAUTHORIZED' }); }); - it('uses an all-false bundle as an access-only gate without projecting dashboard context', async () => { + it('uses an all-false bundle as a Redis-only access gate without projecting dashboard context', async () => { const fixture = buildContext(true, true); const queryRaw = vi.fn(async (_query: unknown) => []); Object.assign(fixture.context.db, { $queryRaw: queryRaw }); @@ -236,12 +234,7 @@ describe('dashboardRouter.getContextBundleDelta', () => { include: { context: false, commandTable: false, boardAccess: false }, }) ).resolves.toEqual({ context: undefined, commandTable: undefined, boardAccess: undefined }); - expect(fixture.findGeneral).toHaveBeenCalledTimes(1); - expect(fixture.findGeneral).toHaveBeenCalledWith({ - where: { userId: auth.user.id }, - orderBy: { id: 'asc' }, - select: { id: true, turnTime: true }, - }); + expect(fixture.findGeneral).not.toHaveBeenCalled(); expect(queryRaw).not.toHaveBeenCalled(); }); @@ -340,8 +333,6 @@ describe('dashboardRouter.getContextBundleDelta', () => { Object.assign(fixture.context.db, { $queryRaw: vi.fn(async () => Promise.reject(new Error('revision table unavailable'))), }); - fixture.context.realtimeAccessGeneralId = 7; - const result = await dashboardRouter.createCaller(fixture.context).getContextBundleDelta({ ...contextOnly, known: { context: 'A'.repeat(22) }, @@ -349,6 +340,6 @@ describe('dashboardRouter.getContextBundleDelta', () => { }); expect(result.context?.kind).toBe('snapshot'); - expect(fixture.findGeneral).toHaveBeenCalledTimes(1); + expect(fixture.findGeneral).toHaveBeenCalledTimes(2); }); }); diff --git a/app/game-api/test/dashboardSourceRevision.test.ts b/app/game-api/test/dashboardSourceRevision.test.ts index 5787bba6..c0b09dc7 100644 --- a/app/game-api/test/dashboardSourceRevision.test.ts +++ b/app/game-api/test/dashboardSourceRevision.test.ts @@ -1,7 +1,10 @@ import { describe, expect, it, vi } from 'vitest'; import type { DatabaseClient } from '../src/context.js'; -import { readDashboardSourceRevisionState } from '../src/services/dashboardSourceRevision.js'; +import { + readDashboardSourceRevisionState, + readDashboardSourceRevisionStateForUser, +} from '../src/services/dashboardSourceRevision.js'; const row = (overrides: Record = {}) => ({ generalId: 7, @@ -13,7 +16,6 @@ const row = (overrides: Record = {}) => ({ cityRevision: 12n, nationRevision: 13n, worldRevision: 14n, - accessRevision: 15n, ...overrides, }); @@ -36,7 +38,6 @@ describe('dashboard source revision', () => { cityRevision: 0n, nationRevision: 0n, worldRevision: 0n, - accessRevision: 0n, }), ]); @@ -47,17 +48,16 @@ describe('dashboard source revision', () => { expect.stringMatching(/^[A-Za-z0-9_-]{22}$/u), ]); const statement = queryRaw.mock.calls[0]?.[0] as { sql: string }; - expect(statement.sql.match(/COALESCE\([^)]*\."revision", 0\)/gu)).toHaveLength(6); + expect(statement.sql.match(/COALESCE\([^)]*\."revision", 0\)/gu)).toHaveLength(5); }); it('hashes exactly the documented context, command, and board dependency vectors', async () => { const initial = (await read([row()])).state; const globalChanged = (await read([row({ globalRevision: 99n })])).state; const cityChanged = (await read([row({ cityRevision: 99n })])).state; - const accessChanged = (await read([row({ accessRevision: 99n })])).state; const worldChanged = (await read([row({ worldRevision: 99n })])).state; const nationChanged = (await read([row({ nationRevision: 99n })])).state; - if (!initial || !globalChanged || !cityChanged || !accessChanged || !worldChanged || !nationChanged) { + if (!initial || !globalChanged || !cityChanged || !worldChanged || !nationChanged) { throw new Error('source revision state missing'); } @@ -67,9 +67,6 @@ describe('dashboard source revision', () => { expect(cityChanged.sourceRevisions.context).not.toBe(initial.sourceRevisions.context); expect(cityChanged.sourceRevisions.commandTable).not.toBe(initial.sourceRevisions.commandTable); expect(cityChanged.sourceRevisions.boardAccess).toBe(initial.sourceRevisions.boardAccess); - expect(accessChanged.sourceRevisions.context).not.toBe(initial.sourceRevisions.context); - expect(accessChanged.sourceRevisions.commandTable).toBe(initial.sourceRevisions.commandTable); - expect(accessChanged.sourceRevisions.boardAccess).toBe(initial.sourceRevisions.boardAccess); expect(worldChanged.sourceRevisions.context).not.toBe(initial.sourceRevisions.context); expect(worldChanged.sourceRevisions.commandTable).not.toBe(initial.sourceRevisions.commandTable); expect(worldChanged.sourceRevisions.boardAccess).toBe(initial.sourceRevisions.boardAccess); @@ -78,6 +75,19 @@ describe('dashboard source revision', () => { expect(nationChanged.sourceRevisions.boardAccess).not.toBe(initial.sourceRevisions.boardAccess); }); + it('resolves the authenticated actor and revision vector in one user-owned statement', async () => { + const queryRaw = vi.fn(async (_query: unknown) => [row()]); + await expect( + readDashboardSourceRevisionStateForUser( + { $queryRaw: queryRaw } as Pick, + 'viewer-7' + ) + ).resolves.toMatchObject({ identity: { generalId: 7 } }); + const statement = queryRaw.mock.calls[0]?.[0] as { sql: string; values: unknown[] }; + expect(statement.sql).toContain('actor."user_id"'); + expect(statement.values).toContain('viewer-7'); + }); + it('includes only the authenticated icon projection in the context source', async () => { const initial = ( await read([row()], { diff --git a/app/game-api/test/deferredGeneralAccess.integration.test.ts b/app/game-api/test/deferredGeneralAccess.integration.test.ts new file mode 100644 index 00000000..466ebc4d --- /dev/null +++ b/app/game-api/test/deferredGeneralAccess.integration.test.ts @@ -0,0 +1,160 @@ +import { randomUUID } from 'node:crypto'; + +import { + createGamePostgresConnector, + createRedisConnector, + resolveRedisConfigFromEnv, + type GamePrismaClient, + type RedisConnector, +} from '@sammo-ts/infra'; +import type { GameSessionTokenPayload } from '@sammo-ts/common/auth/gameToken'; +import { afterAll, beforeAll, describe, expect, it } from 'vitest'; + +import { + DeferredGeneralAccessWorker, + enqueueDeferredGeneralAccess, + getDeferredGeneralAccessLimit, +} from '../src/services/deferredGeneralAccess.js'; + +const databaseUrl = process.env.INPUT_EVENT_DATABASE_URL; +const liveIntegration = describe.skipIf(!databaseUrl || !process.env.REDIS_URL); + +liveIntegration('deferred general access with PostgreSQL and Redis', () => { + const runId = randomUUID(); + const profile = `deferred-access-${runId}`; + const userId = `deferred-user-${runId}`; + const scenarioCode = `deferred-scenario-${runId}`; + const batchId = `deferred-batch-${runId}`; + const generalId = 9_981_000 + Math.floor(Math.random() * 900); + const now = new Date('2026-08-17T03:05:00.000Z'); + const nextAccessAt = new Date('2026-08-17T03:10:00.000Z'); + const auth: GameSessionTokenPayload = { + version: 1, + profile, + issuedAt: '2026-08-17T03:00:00.000Z', + expiresAt: '2026-08-18T03:00:00.000Z', + sessionId: `deferred-session-${runId}`, + user: { + id: userId, + username: userId, + displayName: userId, + roles: [], + }, + sanctions: {}, + }; + let db: GamePrismaClient; + let closeDb: (() => Promise) | undefined; + let redis: RedisConnector; + let worldStateId: number; + + beforeAll(async () => { + const dbConnector = createGamePostgresConnector({ url: databaseUrl! }); + await dbConnector.connect(); + db = dbConnector.prisma; + closeDb = () => dbConnector.disconnect(); + redis = createRedisConnector(resolveRedisConfigFromEnv()); + await redis.connect(); + const world = await db.worldState.create({ + data: { + scenarioCode, + currentYear: 200, + currentMonth: 1, + tickSeconds: 600, + config: {}, + meta: { + lastTurnTime: '2026-08-17T03:00:00.000Z', + refreshLimit: 1, + }, + }, + }); + worldStateId = world.id; + await db.general.create({ + data: { + id: generalId, + userId, + name: '지연접속검증', + turnTime: nextAccessAt, + }, + }); + }); + + afterAll(async () => { + if (redis) { + const keys: string[] = []; + for await (const page of redis.client.scanIterator({ + MATCH: `sammo:game:general-access:*:${profile}*`, + COUNT: 100, + })) { + keys.push(...page); + } + if (keys.length > 0) await redis.client.del(keys); + await redis.disconnect(); + } + if (db) { + await db.generalAccessBatch.deleteMany({ where: { id: batchId } }); + await db.generalAccessLog.deleteMany({ where: { generalId } }); + await db.general.deleteMany({ where: { id: generalId } }); + await db.worldState.deleteMany({ where: { id: worldStateId } }); + await closeDb?.(); + } + }); + + it('aggregates Redis increments, flushes once, and installs a lazy limit marker', async () => { + const workerDb = { + worldState: { + findFirst: () => + db.worldState.findUniqueOrThrow({ + where: { id: worldStateId }, + select: { + id: true, + currentYear: true, + currentMonth: true, + tickSeconds: true, + meta: true, + }, + }), + }, + $queryRaw: db.$queryRaw.bind(db), + $executeRaw: db.$executeRaw.bind(db), + }; + const worker = new DeferredGeneralAccessWorker( + workerDb as never, + redis.client, + profile, + { get: async () => 'RUNNING' }, + { now: () => now, createBatchId: () => batchId } + ); + + await enqueueDeferredGeneralAccess(redis.client, profile, auth, generalId, 1, now); + await enqueueDeferredGeneralAccess(redis.client, profile, auth, generalId, 1, now); + await expect(db.generalAccessLog.findUnique({ where: { generalId } })).resolves.toBeNull(); + + await worker.flushOnce(); + await expect(db.generalAccessLog.findUniqueOrThrow({ where: { generalId } })).resolves.toMatchObject({ + userId, + refresh: 2, + refreshTotal: 2, + refreshScore: 2, + refreshScoreTotal: 2, + }); + await expect( + db.trafficPeriod.findUniqueOrThrow({ + where: { + worldStateId_year_month: { + worldStateId, + year: 200, + month: 1, + }, + }, + }) + ).resolves.toMatchObject({ refresh: 2, online: 1 }); + await expect(getDeferredGeneralAccessLimit(redis.client, profile, auth, now)).resolves.toEqual({ + nextAccessAt, + }); + + await worker.flushOnce(); + await expect(db.generalAccessLog.findUniqueOrThrow({ where: { generalId } })).resolves.toMatchObject({ + refreshTotal: 2, + }); + }); +}); diff --git a/app/game-api/test/deferredGeneralAccess.test.ts b/app/game-api/test/deferredGeneralAccess.test.ts new file mode 100644 index 00000000..5d18b6d9 --- /dev/null +++ b/app/game-api/test/deferredGeneralAccess.test.ts @@ -0,0 +1,140 @@ +import { describe, expect, it, vi } from 'vitest'; + +import type { GameSessionTokenPayload } from '@sammo-ts/common/auth/gameToken'; + +import { + DeferredGeneralAccessWorker, + enqueueDeferredGeneralAccess, + flushDeferredGeneralAccessBatch, + getDeferredGeneralAccessLimit, +} from '../src/services/deferredGeneralAccess.js'; + +const auth = (roles: string[] = []): GameSessionTokenPayload => ({ + version: 1, + profile: 'hwe:default', + issuedAt: '2026-08-17T00:00:00.000Z', + expiresAt: '2026-08-18T00:00:00.000Z', + sessionId: 'deferred-access-session', + user: { + id: 'deferred-user', + username: 'deferred-user', + displayName: '지연 접속 사용자', + roles, + }, + sanctions: {}, +}); + +describe('deferred general access', () => { + it('adds a server-owned access weight with one Redis script and skips admins', async () => { + const evalScript = vi.fn( + async (_script: string, _options: { keys: string[]; arguments: string[] }) => 1 + ); + const now = new Date('2026-08-17T03:00:00.000Z'); + + await expect( + enqueueDeferredGeneralAccess({ eval: evalScript }, 'hwe:default', auth(), 7, 1, now) + ).resolves.toBe(true); + expect(evalScript).toHaveBeenCalledTimes(1); + expect(evalScript.mock.calls[0]?.[1]).toMatchObject({ + keys: ['sammo:game:general-access:pending:hwe:default'], + arguments: ['7', 'deferred-user', '1', String(now.getTime()), String(24 * 60 * 60 * 1_000)], + }); + + await expect( + enqueueDeferredGeneralAccess({ eval: evalScript }, 'hwe:default', auth(['admin']), 7, 1, now) + ).resolves.toBe(false); + expect(evalScript).toHaveBeenCalledTimes(1); + }); + + it('reads only an unexpired Redis limit marker', async () => { + const nextAccessAt = '2026-08-17T03:10:00.000Z'; + await expect( + getDeferredGeneralAccessLimit( + { get: vi.fn(async () => JSON.stringify({ nextAccessAt })) }, + 'hwe:default', + auth(), + new Date('2026-08-17T03:05:00.000Z') + ) + ).resolves.toEqual({ nextAccessAt: new Date(nextAccessAt) }); + await expect( + getDeferredGeneralAccessLimit( + { get: vi.fn(async () => JSON.stringify({ nextAccessAt })) }, + 'hwe:default', + auth(), + new Date(nextAccessAt) + ) + ).resolves.toBeNull(); + }); + + it('flushes an aggregated batch with one PostgreSQL write statement and no read-model journal', async () => { + const queryRaw = vi.fn(async (_query: unknown) => [ + { + generalId: 7, + userId: 'deferred-user', + refreshScore: 2, + nextAccessAt: new Date('2026-08-17T03:10:00.000Z'), + }, + ]); + const db = { + worldState: { + findFirst: vi.fn(async () => ({ + id: 1, + currentYear: 200, + currentMonth: 1, + tickSeconds: 600, + meta: { lastTurnTime: '2026-08-17T03:00:00.000Z', refreshLimit: 50 }, + })), + }, + $queryRaw: queryRaw, + $executeRaw: vi.fn(async () => 1), + }; + + await expect( + flushDeferredGeneralAccessBatch(db as never, 'batch-1', [ + { + generalId: 7, + userId: 'deferred-user', + weight: 2, + lastRefresh: new Date('2026-08-17T03:05:00.000Z'), + }, + ]) + ).resolves.toMatchObject({ refreshLimit: 50, states: [{ refreshScore: 2 }] }); + + expect(queryRaw).toHaveBeenCalledTimes(1); + const statement = queryRaw.mock.calls[0]?.[0] as { sql: string }; + expect(statement.sql).toContain('INSERT INTO "general_access_batch"'); + expect(statement.sql).toContain('jsonb_to_recordset'); + expect(statement.sql).toContain('INSERT INTO "traffic_period"'); + expect(statement.sql).toContain('INSERT INTO "traffic_period_general"'); + expect(statement.sql).toContain('INSERT INTO "general_access_log"'); + expect(statement.sql).not.toContain('read_model_revision'); + expect(statement.sql).not.toContain('read_model_outbox'); + }); + + it('drops a rotated batch when profile status is unavailable instead of penalizing it later', async () => { + const del = vi.fn(async (_key: string) => 1); + const hGetAll = vi.fn(async (_key: string) => ({})); + const redis = { + scanIterator: async function* (_options: { MATCH: string; COUNT: number }) { + yield []; + }, + eval: vi.fn(async () => 1), + hGetAll, + get: vi.fn(async () => null), + set: vi.fn(async () => 'OK'), + del, + }; + const worker = new DeferredGeneralAccessWorker( + {} as never, + redis, + 'hwe:default', + { get: async () => Promise.reject(new Error('gateway unavailable')) }, + { createBatchId: () => 'status-failure-batch' } + ); + + await worker.flushOnce(); + + expect(del).toHaveBeenCalledWith('sammo:game:general-access:batch:hwe:default:status-failure-batch'); + expect(hGetAll).not.toHaveBeenCalled(); + }); +}); diff --git a/app/game-api/test/generalAccessTracking.integration.test.ts b/app/game-api/test/generalAccessTracking.integration.test.ts index 3f7b06b7..f3821e3d 100644 --- a/app/game-api/test/generalAccessTracking.integration.test.ts +++ b/app/game-api/test/generalAccessTracking.integration.test.ts @@ -1,4 +1,4 @@ -import { afterAll, beforeAll, describe, expect, it } from 'vitest'; +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; import { TRPCError } from '@trpc/server'; import type { GameSessionTokenPayload } from '@sammo-ts/common/auth/gameToken'; import { createGamePostgresConnector, type GamePrismaClient } from '@sammo-ts/infra'; @@ -7,6 +7,7 @@ import { z } from 'zod'; import type { GameApiContext } from '../src/context.js'; import { appRouter } from '../src/router.js'; import { upsertGeneralAccess } from '../src/services/generalAccess.js'; +import { flushDeferredGeneralAccessBatch } from '../src/services/deferredGeneralAccess.js'; import { accessAuthedInputProcedure, router } from '../src/trpc.js'; const databaseUrl = process.env.INPUT_EVENT_DATABASE_URL; @@ -20,6 +21,7 @@ const endpointUserId = `access-endpoint-user-${endpointGeneralId}`; const scenarioCode = `traffic-period-${generalId}`; const endpointRequestPrefix = `access-endpoint-${endpointGeneralId}`; const yearbookProfile = `access-profile-${endpointGeneralId}`; +const deferredBatchId = `deferred-access-${endpointGeneralId}`; const endpointAuth = (roles = ['user']): GameSessionTokenPayload => ({ version: 1, @@ -84,6 +86,7 @@ integration('general access tracking persistence', () => { }, }, }); + await db.generalAccessBatch.deleteMany({ where: { id: deferredBatchId } }); await db.inputEvent.deleteMany({ where: { requestId: { startsWith: endpointRequestPrefix } } }); await db.worldState.deleteMany({ where: { scenarioCode } }); const world = await db.worldState.create({ @@ -135,6 +138,7 @@ integration('general access tracking persistence', () => { }, }, }); + await db.generalAccessBatch.deleteMany({ where: { id: deferredBatchId } }); await db.inputEvent.deleteMany({ where: { requestId: { startsWith: endpointRequestPrefix } } }); await db.yearbookHistory.deleteMany({ where: { profileName: yearbookProfile } }); await db.general.deleteMany({ where: { id: endpointGeneralId } }); @@ -390,6 +394,7 @@ integration('general access tracking persistence', () => { }); it('runs parser, access, business event, zero-weight, admin, and yearbook cache boundaries on PostgreSQL', async () => { + const redisEval = vi.fn(async () => 1); const context = { auth: endpointAuth(), db, @@ -404,6 +409,7 @@ integration('general access tracking persistence', () => { redis: { get: async () => null, set: async () => 'OK', + eval: redisEval, }, } as unknown as GameApiContext; const boundaryCaller = endpointBoundaryRouter.createCaller(context); @@ -419,8 +425,21 @@ integration('general access tracking persistence', () => { expect(initialDashboard).toMatchObject({ context: { kind: 'snapshot' } }); const initialContext = initialDashboard.context; if (!initialContext?.sourceRevision) { - throw new Error('dashboard snapshot did not include its post-access source revision'); + throw new Error('dashboard snapshot did not include its source revision'); } + await expect( + db.generalAccessLog.findUnique({ where: { generalId: endpointGeneralId } }) + ).resolves.toBeNull(); + expect(redisEval).toHaveBeenCalledTimes(1); + + await flushDeferredGeneralAccessBatch(db, deferredBatchId, [ + { + generalId: endpointGeneralId, + userId: endpointUserId, + weight: 1, + lastRefresh: new Date('2026-07-26T02:55:00.000Z'), + }, + ]); await expect( db.generalAccessLog.findUniqueOrThrow({ where: { generalId: endpointGeneralId } }) ).resolves.toMatchObject({ refresh: 1, refreshTotal: 1 }); diff --git a/app/game-api/test/generalAccessTracking.test.ts b/app/game-api/test/generalAccessTracking.test.ts index c396b45e..57310638 100644 --- a/app/game-api/test/generalAccessTracking.test.ts +++ b/app/game-api/test/generalAccessTracking.test.ts @@ -5,7 +5,12 @@ import { z } from 'zod'; import type { GameApiContext } from '../src/context.js'; import type { DatabaseClient } from '../src/context.js'; -import { accessAuthedInputProcedure, accessLimitAuthedProcedure, router } from '../src/trpc.js'; +import { + accessAuthedInputProcedure, + accessLimitAuthedProcedure, + deferredAccessLimitAuthedProcedure, + router, +} from '../src/trpc.js'; import { accessPageWeights, generalAccessEndpointWeights, @@ -164,7 +169,7 @@ describe('general access tracking', () => { select: { id: true, userId: true, turnTime: true }, }); expect(transaction).toHaveBeenCalledTimes(1); - expect(queryRaw).toHaveBeenCalledTimes(2); + expect(queryRaw).toHaveBeenCalledTimes(1); expect(executeRaw).toHaveBeenCalledTimes(2); const periodStatement = queryRaw.mock.calls[0]![0] as { sql: string; values: unknown[] }; @@ -193,10 +198,6 @@ describe('general access tracking', () => { expect(accessStatement.values).toContain(now); expect(accessStatement.values).toContainEqual(new Date('2026-07-26T03:00:00.000Z')); - const journalStatement = queryRaw.mock.calls[1]![0] as { sql: string; values: unknown[] }; - expect(journalStatement.sql).toContain('INSERT INTO "read_model_outbox"'); - expect(journalStatement.values).toContain('access.general'); - expect(journalStatement.values).toContain(7); }); it('accepts legacy weight zero to refresh timestamps without incrementing counters', async () => { @@ -243,6 +244,32 @@ describe('general access tracking', () => { expect(resolver).not.toHaveBeenCalled(); }); + it('checks a deferred limit with Redis only and does not query PostgreSQL', async () => { + const fixture = buildDb(); + const resolver = vi.fn(() => ({ ok: true })); + const limitedRouter = router({ read: deferredAccessLimitAuthedProcedure.query(resolver) }); + const get = vi.fn(async () => + JSON.stringify({ nextAccessAt: '2099-07-26T03:10:00.000Z' }) + ); + + await expect( + limitedRouter + .createCaller({ + ...accessContext(fixture.db), + redis: { get }, + } as unknown as GameApiContext) + .read() + ).rejects.toMatchObject({ + code: 'TOO_MANY_REQUESTS', + message: expect.stringContaining('다음 접속 가능 시각'), + }); + expect(get).toHaveBeenCalledTimes(1); + expect(fixture.findGeneral).not.toHaveBeenCalled(); + expect(fixture.findWorld).not.toHaveBeenCalled(); + expect(fixture.db.generalAccessLog.findUnique).not.toHaveBeenCalled(); + expect(resolver).not.toHaveBeenCalled(); + }); + it('rejects weights that cannot come from a server-owned Ref call boundary', async () => { const fixture = buildDb(); await expect(recordGeneralAccessWeight(accessContext(fixture.db), -1)).rejects.toBeInstanceOf(RangeError); diff --git a/packages/infra/prisma/game.prisma b/packages/infra/prisma/game.prisma index 56fb603f..df8598f8 100644 --- a/packages/infra/prisma/game.prisma +++ b/packages/infra/prisma/game.prisma @@ -280,6 +280,14 @@ model GeneralAccessLog { @@map("general_access_log") } +model GeneralAccessBatch { + id String @id @db.VarChar(64) + createdAt DateTime @default(now()) @map("created_at") + + @@index([createdAt]) + @@map("general_access_batch") +} + model TrafficPeriod { id Int @id @default(autoincrement()) worldStateId Int @map("world_state_id") diff --git a/packages/infra/prisma/migrations/20260817000000_add_general_access_batch/migration.sql b/packages/infra/prisma/migrations/20260817000000_add_general_access_batch/migration.sql new file mode 100644 index 00000000..d38f8741 --- /dev/null +++ b/packages/infra/prisma/migrations/20260817000000_add_general_access_batch/migration.sql @@ -0,0 +1,9 @@ +CREATE TABLE "general_access_batch" ( + "id" VARCHAR(64) NOT NULL, + "created_at" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + + CONSTRAINT "general_access_batch_pkey" PRIMARY KEY ("id") +); + +CREATE INDEX "general_access_batch_created_at_idx" + ON "general_access_batch"("created_at"); diff --git a/packages/infra/prisma/migrations/README.md b/packages/infra/prisma/migrations/README.md index acc8d17e..e3549f19 100644 --- a/packages/infra/prisma/migrations/README.md +++ b/packages/infra/prisma/migrations/README.md @@ -35,6 +35,7 @@ chain을 적용하고 두 번째 실행은 `No pending migrations to apply`여 - `city.trade` nullable, `city.trust` REAL - `auction_bid.meta` JSONB NOT NULL - `traffic_period`, `traffic_period_general`과 unique key +- `general_access_batch`, primary key와 `created_at` index - `select_npc_token`, `select_npc_token_valid_until_idx` - `general_user_id_key`