fix(housekeeping): complete people workflow fidelity

This commit is contained in:
Simo committed 2026-08-29 14:39:38 +02:00
1 parent 25b76437ff
commit 91c9efcbb4
29 files changed
+1365 -290

No files matched your search

@@ -1,7 +1,7 @@
import "server-only";
import crypto from "node:crypto";
import { and, eq, inArray, max, sql } from "drizzle-orm";
import { and, desc, eq, gt, inArray, max, or, sql } from "drizzle-orm";
import type { ResultSetHeader } from "mysql2";
import { AuditOutcomePersistenceError } from "@/features/housekeeping/foundation/commands/audit-envelope";
import { invalidateLoginCache } from "@/lib/auth";
@@ -41,6 +41,7 @@ import {
fail,
type HousekeepingCapabilityContext,
type HousekeepingErrorCode,
type HousekeepingPartialCompletion,
type HousekeepingResult,
ok,
} from "../../../foundation/contracts";
@@ -72,6 +73,7 @@ export interface PeopleMutationSnapshot {
readonly before: Readonly<Record<string, unknown>> | null;
readonly after: Readonly<Record<string, unknown>> | null;
readonly output?: Readonly<Record<string, unknown>>;
readonly completion?: HousekeepingPartialCompletion;
}
export interface PeopleMutationInvocation {
@@ -164,16 +166,22 @@ export function createPeopleMutationService(
legacy: invocation.legacy === true,
};
try {
return ok(
await adapter.execute(operation, input, context),
invocation.correlationId,
);
const snapshot = await adapter.execute(operation, input, context);
return ok(snapshot, invocation.correlationId, snapshot.completion);
} catch (error) {
if (error instanceof AuditOutcomePersistenceError) {
if (invocation.legacy) {
return ok(error.operationResult, invocation.correlationId);
}
throw error;
const operationSnapshot =
error.operationResult as PeopleMutationSnapshot;
const completion: HousekeepingPartialCompletion = {
status: "partial",
external: operationSnapshot.completion?.external ?? "completed",
audit: "unavailable",
};
const snapshot = {
...operationSnapshot,
completion,
};
return ok(snapshot, invocation.correlationId, completion);
}
if (error instanceof PeopleMutationFailure) {
return fail(
@@ -221,6 +229,43 @@ function positiveInteger(value: unknown): number {
return parsed;
}
const MAX_UNSIGNED_BIGINT = 18_446_744_073_709_551_615n;
const MAX_AUDIT_TARGET_ID = 2_147_483_647n;
function positiveBigIntIdentifier(value: unknown): bigint {
let parsed: bigint;
if (typeof value === "bigint") {
parsed = value;
} else if (typeof value === "string" && /^[1-9]\d{0,19}$/u.test(value)) {
parsed = BigInt(value);
} else if (
typeof value === "number" &&
Number.isSafeInteger(value) &&
value > 0
) {
parsed = BigInt(value);
} else {
throw new PeopleMutationFailure(
"VALIDATION",
"errors.housekeeping.validation",
);
}
if (parsed <= 0n || parsed > MAX_UNSIGNED_BIGINT) {
throw new PeopleMutationFailure(
"VALIDATION",
"errors.housekeeping.validation",
);
}
return parsed;
}
function serializeBigIntIdentifier(value: unknown): string {
return positiveBigIntIdentifier(value).toString();
}
function auditTargetId(value: bigint): number | undefined {
return value <= MAX_AUDIT_TARGET_ID ? Number(value) : undefined;
}
function nonNegativeInteger(value: unknown): number {
const parsed = Number(value);
if (!Number.isSafeInteger(parsed) || parsed < 0) {
@@ -356,10 +401,12 @@ async function finalizeExternalWithAudit(
targetId: number | undefined,
snapshot: PeopleMutationSnapshot,
execute: () => Promise<void>,
mutationCommitted = true,
): Promise<PeopleMutationSnapshot> {
try {
await execute();
} catch (error) {
let audit: HousekeepingPartialCompletion["audit"] = "persisted";
try {
await logAudit(
canonicalAuditEntry(
@@ -368,13 +415,23 @@ async function finalizeExternalWithAudit(
target,
targetId,
snapshot,
"failure",
mutationCommitted ? "partial" : "failure",
),
);
} catch {
// Preserve the external failure as the authoritative result.
audit = "unavailable";
}
throw error;
if (
!mutationCommitted ||
(context.legacy && !(error instanceof PeopleMutationFailure))
) {
throw error;
}
return withPartialCompletion(snapshot, {
status: "partial",
external: "failed",
audit,
});
}
try {
await logAudit(
@@ -387,7 +444,9 @@ async function finalizeExternalWithAudit(
"success",
),
);
} catch (auditOutcomeError) {
return snapshot;
} catch {
let audit: HousekeepingPartialCompletion["audit"] = "persisted";
try {
await logAudit(
canonicalAuditEntry(
@@ -400,13 +459,22 @@ async function finalizeExternalWithAudit(
),
);
} catch {
// The typed completed-operation error remains primary.
audit = "unavailable";
}
throw new AuditOutcomePersistenceError(snapshot, auditOutcomeError);
return withPartialCompletion(snapshot, {
status: "partial",
external: "completed",
audit,
});
}
return snapshot;
}
function withPartialCompletion(
snapshot: PeopleMutationSnapshot,
completion: HousekeepingPartialCompletion,
): PeopleMutationSnapshot {
return { ...snapshot, completion };
}
async function runExternalWithAudit(
context: PeopleMutationContext,
operation: PeopleMutationOperation,
@@ -432,6 +500,7 @@ async function runExternalWithAudit(
targetId,
snapshot,
execute,
false,
);
}
async function executeUserMutation(
@@ -441,8 +510,24 @@ async function executeUserMutation(
): Promise<PeopleMutationSnapshot> {
const data = record(input);
const userId = positiveInteger(data.userId);
const guardHierarchy =
operation !== "user.alert" && operation !== "user.trade-lock";
if (operation === "user.alert") {
const message = normalizedText(data.message, 500);
const snapshot: PeopleMutationSnapshot = {
before: null,
after: { userId, alertDelivered: true },
};
return runExternalWithAudit(
context,
operation,
"User",
userId,
snapshot,
async () => requireRcon(await rcon.alertUser(userId, message)),
);
}
const guardHierarchy = operation !== "user.trade-lock";
const target = await loadTarget(userId, context, guardHierarchy);
const before = targetSnapshot(target);
@@ -494,19 +579,28 @@ async function executeUserMutation(
"User",
userId,
snapshot,
"success",
"intent",
),
tx,
);
});
invalidateLoginCache(target.username);
await notify({
action: "user_edit",
actor: context.capability.actor.username,
target: target.username,
targetId: userId,
});
return snapshot;
return finalizeExternalWithAudit(
context,
operation,
"User",
userId,
snapshot,
async () => {
invalidateLoginCache(target.username);
const notification = notify({
action: "user_edit",
actor: context.capability.actor.username,
target: target.username,
targetId: userId,
});
if (!context.legacy) await notification;
},
);
}
if (operation === "user.ban") {
@@ -532,7 +626,13 @@ async function executeUserMutation(
type: Ban.type,
})
.from(Ban)
.where(eq(Ban.userId, userId))
.where(
and(
eq(Ban.userId, userId),
or(eq(Ban.banExpire, 0), gt(Ban.banExpire, now)),
),
)
.orderBy(desc(Ban.timestamp), desc(Ban.id))
.limit(1);
await tx.insert(Ban).values({
userId,
@@ -544,6 +644,22 @@ async function executeUserMutation(
ip: normalizedText(data.ip, 255, false),
machineId: "",
});
const [observedAfterBan] = await tx
.select({
id: Ban.id,
banExpire: Ban.banExpire,
banReason: Ban.banReason,
type: Ban.type,
})
.from(Ban)
.where(
and(
eq(Ban.userId, userId),
or(eq(Ban.banExpire, 0), gt(Ban.banExpire, now)),
),
)
.orderBy(desc(Ban.timestamp), desc(Ban.id))
.limit(1);
snapshot = {
before: {
...before,
@@ -556,7 +672,17 @@ async function executeUserMutation(
}
: {}),
},
after: { ...before, banned: true, banExpire, type, reason },
after: {
...before,
banned: observedAfterBan !== undefined,
...(observedAfterBan
? {
banExpire: observedAfterBan.banExpire,
type: observedAfterBan.type,
reason: observedAfterBan.banReason,
}
: {}),
},
};
await logAudit(
canonicalAuditEntry(
@@ -577,7 +703,7 @@ async function executeUserMutation(
userId,
snapshot,
async () => {
await rcon.disconnectUser(userId);
await requireRcon(await rcon.disconnectUser(userId));
await notify({
action: "ban",
actor: context.capability.actor.username,
@@ -589,6 +715,7 @@ async function executeUserMutation(
}
if (operation === "user.unban") {
const now = Math.floor(Date.now() / 1000);
const snapshot = await db.transaction(async (tx) => {
const [activeBan] = await tx
.select({
@@ -598,9 +725,31 @@ async function executeUserMutation(
type: Ban.type,
})
.from(Ban)
.where(eq(Ban.userId, userId))
.where(
and(
eq(Ban.userId, userId),
or(eq(Ban.banExpire, 0), gt(Ban.banExpire, now)),
),
)
.orderBy(desc(Ban.timestamp), desc(Ban.id))
.limit(1);
await tx.delete(Ban).where(eq(Ban.userId, userId));
const [observedAfterBan] = await tx
.select({
id: Ban.id,
banExpire: Ban.banExpire,
banReason: Ban.banReason,
type: Ban.type,
})
.from(Ban)
.where(
and(
eq(Ban.userId, userId),
or(eq(Ban.banExpire, 0), gt(Ban.banExpire, now)),
),
)
.orderBy(desc(Ban.timestamp), desc(Ban.id))
.limit(1);
const value: PeopleMutationSnapshot = {
before: {
...before,
@@ -613,7 +762,17 @@ async function executeUserMutation(
}
: {}),
},
after: { ...before, banned: false },
after: {
...before,
banned: observedAfterBan !== undefined,
...(observedAfterBan
? {
banExpire: observedAfterBan.banExpire,
type: observedAfterBan.type,
reason: observedAfterBan.banReason,
}
: {}),
},
};
await logAudit(
canonicalAuditEntry(
@@ -636,19 +795,6 @@ async function executeUserMutation(
return snapshot;
}
if (operation === "user.alert") {
const message = normalizedText(data.message, 500);
const snapshot = { before, after: { ...before, alertDelivered: true } };
return runExternalWithAudit(
context,
operation,
"User",
userId,
snapshot,
async () => requireRcon(await rcon.alertUser(userId, message)),
);
}
if (operation === "user.disconnect") {
const snapshot = { before, after: { ...before, online: false } };
return runExternalWithAudit(
@@ -718,19 +864,28 @@ async function executeUserMutation(
"User",
userId,
snapshot,
"success",
"intent",
),
tx,
);
});
invalidateLoginCache(target.username);
await notify({
action: "user_edit",
actor: context.capability.actor.username,
target: target.username,
details: "Password reset",
});
return snapshot;
return finalizeExternalWithAudit(
context,
operation,
"User",
userId,
snapshot,
async () => {
invalidateLoginCache(target.username);
const notification = notify({
action: "user_edit",
actor: context.capability.actor.username,
target: target.username,
details: "Password reset",
});
if (!context.legacy) await notification;
},
);
}
if (operation === "user.send-currency") {
@@ -801,8 +956,10 @@ async function executeUserMutation(
after: {
...before,
tradeLockedUntil: untilUnix,
canTrade: locked ? "0" : "1",
tradelockAmount: (settings?.tradelockAmount ?? 0) + (locked ? 1 : 0),
canTrade: settings ? (locked ? "0" : "1") : null,
tradelockAmount: settings
? settings.tradelockAmount + (locked ? 1 : 0)
: null,
},
output: { userId, untilUnix },
};
@@ -825,15 +982,17 @@ async function executeUserMutation(
userId,
snapshot,
async () => {
await rcon.setTradeLock(userId, locked);
await rcon.alertUser(
userId,
locked
? "Trading has been disabled by staff."
: "Trading has been re-enabled by staff.",
await requireRcon(await rcon.setTradeLock(userId, locked));
await requireRcon(
await rcon.alertUser(
userId,
locked
? "Trading has been disabled by staff."
: "Trading has been re-enabled by staff.",
),
);
if (target.online === "1") {
await rcon.disconnectUser(userId, target.username);
await requireRcon(await rcon.disconnectUser(userId, target.username));
}
await logStaffActivity({
staffId: context.capability.actor.id,
@@ -856,6 +1015,7 @@ async function executeBulkMutation(
const ids = userIds(data.userIds, context.legacy);
const failures: Array<{ userId: number; reason: string }> = [];
let completed = 0;
let externalFailed = false;
let mutationReason: string | undefined;
if (operation === "users.bulk-unban") {
@@ -966,13 +1126,16 @@ async function executeBulkMutation(
| "pixels"
| "points";
for (const userId of ids) {
let databaseCommitted = false;
try {
if (type === "credits") {
await db
.update(User)
.set({ credits: sql`${User.credits} + ${amount}` })
.where(eq(User.id, userId));
await rcon.giveCredits(userId, amount);
databaseCommitted = true;
const delivered = await rcon.giveCredits(userId, amount);
if (!context.legacy) await requireRcon(delivered);
} else {
const currencyType = type === "pixels" ? 0 : 101;
await db
@@ -981,11 +1144,18 @@ async function executeBulkMutation(
.onDuplicateKeyUpdate({
set: { amount: sql`${UsersCurrency.amount} + ${amount}` },
});
if (type === "pixels") await rcon.giveDuckets(userId, amount);
else await rcon.givePointsGotw(userId, amount);
databaseCommitted = true;
if (type === "pixels") {
const delivered = await rcon.giveDuckets(userId, amount);
if (!context.legacy) await requireRcon(delivered);
} else {
const delivered = await rcon.givePointsGotw(userId, amount);
if (!context.legacy) await requireRcon(delivered);
}
}
completed += 1;
} catch {
if (databaseCommitted) externalFailed = true;
failures.push({ userId, reason: "Database error" });
}
}
@@ -998,6 +1168,7 @@ async function executeBulkMutation(
} else {
const badgeCode = normalizedText(data.badgeCode, 20);
for (const userId of ids) {
let databaseCommitted = false;
try {
const [existing] = await db
.select({ id: UsersBadges.id })
@@ -1019,10 +1190,13 @@ async function executeBulkMutation(
slotId: (aggregate?.maxSlot ?? 0) + 1,
badgeCode,
});
await rcon.giveBadge(userId, badgeCode);
databaseCommitted = true;
const delivered = await rcon.giveBadge(userId, badgeCode);
if (!context.legacy) await requireRcon(delivered);
}
completed += 1;
} catch {
if (databaseCommitted) externalFailed = true;
failures.push({ userId, reason: "Database error" });
}
}
@@ -1034,6 +1208,14 @@ async function executeBulkMutation(
});
}
const completion: HousekeepingPartialCompletion | undefined =
failures.length > 0
? {
status: "partial",
external: externalFailed ? "failed" : "not-required",
audit: "persisted",
}
: undefined;
const snapshot: PeopleMutationSnapshot = {
before: { userIds: ids },
after: {
@@ -1042,6 +1224,7 @@ async function executeBulkMutation(
failedIds: failures,
...(mutationReason ? { reason: mutationReason } : {}),
},
...(completion === undefined ? {} : { completion }),
};
try {
await logAudit(
@@ -1137,7 +1320,7 @@ async function executeApplicationDecision(
context: PeopleMutationContext,
): Promise<PeopleMutationSnapshot> {
const data = record(input);
const applicationId = positiveInteger(data.applicationId);
const applicationId = positiveBigIntIdentifier(data.applicationId);
if (data.decision !== "dismiss") {
throw new PeopleMutationFailure(
"VALIDATION",
@@ -1153,7 +1336,7 @@ async function executeApplicationDecision(
content: WebsiteStaffApplications.content,
})
.from(WebsiteStaffApplications)
.where(eq(WebsiteStaffApplications.id, BigInt(applicationId)))
.where(eq(WebsiteStaffApplications.id, applicationId))
.limit(1);
if (!application) {
throw new PeopleMutationFailure(
@@ -1163,10 +1346,10 @@ async function executeApplicationDecision(
}
await tx
.delete(WebsiteStaffApplications)
.where(eq(WebsiteStaffApplications.id, BigInt(applicationId)));
.where(eq(WebsiteStaffApplications.id, applicationId));
const snapshot = {
before: {
id: Number(application.id),
id: serializeBigIntIdentifier(application.id),
userId: application.userId,
rankId: application.rankId,
content: application.content,
@@ -1178,7 +1361,7 @@ async function executeApplicationDecision(
context,
"application.decide",
"StaffApplication",
applicationId,
auditTargetId(applicationId),
snapshot,
"success",
),
@@ -1194,7 +1377,7 @@ async function executeTeamChange(
): Promise<PeopleMutationSnapshot> {
const data = record(input);
if (data.action === "delete") {
const teamId = positiveInteger(data.teamId);
const teamId = positiveBigIntIdentifier(data.teamId);
return db.transaction(async (tx) => {
const [team] = await tx
.select({
@@ -1205,7 +1388,7 @@ async function executeTeamChange(
hiddenRank: WebsiteTeams.hiddenRank,
})
.from(WebsiteTeams)
.where(eq(WebsiteTeams.id, BigInt(teamId)))
.where(eq(WebsiteTeams.id, teamId))
.limit(1);
if (!team) {
throw new PeopleMutationFailure(
@@ -1213,9 +1396,9 @@ async function executeTeamChange(
"errors.housekeeping.notFound",
);
}
await tx.delete(WebsiteTeams).where(eq(WebsiteTeams.id, BigInt(teamId)));
await tx.delete(WebsiteTeams).where(eq(WebsiteTeams.id, teamId));
const snapshot: PeopleMutationSnapshot = {
before: { ...team, id: Number(team.id) },
before: { ...team, id: serializeBigIntIdentifier(team.id) },
after: null,
};
await logAudit(
@@ -1223,7 +1406,7 @@ async function executeTeamChange(
context,
"team.change",
"Team",
teamId,
auditTargetId(teamId),
snapshot,
"success",
),
@@ -1249,11 +1432,11 @@ async function executeTeamChange(
createdAt: now,
updatedAt: now,
});
const teamId = Number(result.insertId);
const teamId = positiveBigIntIdentifier(result.insertId);
const snapshot: PeopleMutationSnapshot = {
before: null,
after: {
teamId,
teamId: teamId.toString(),
rankName,
badge: badge || null,
jobDescription: jobDescription || null,
@@ -1266,7 +1449,7 @@ async function executeTeamChange(
context,
"team.change",
"Team",
teamId,
auditTargetId(teamId),
snapshot,
"success",
),
@@ -1309,17 +1492,17 @@ async function executeIpAction(
: await tx
.insert(WebsiteIpWhitelist)
.values({ ipAddress, asn, whitelistAsn: asn !== null });
const id = Number(result.insertId);
const id = positiveBigIntIdentifier(result.insertId);
const snapshot: PeopleMutationSnapshot = {
before: null,
after: { id, action, ipAddress, asn },
after: { id: id.toString(), action, ipAddress, asn },
};
await logAudit(
canonicalAuditEntry(
context,
"ip.action",
"IpRule",
id,
auditTargetId(id),
snapshot,
"success",
),
@@ -1328,7 +1511,7 @@ async function executeIpAction(
return snapshot;
});
}
const id = positiveInteger(data.id);
const id = positiveBigIntIdentifier(data.id);
return db.transaction(async (tx) => {
const [existing] = blacklist
? await tx
@@ -1338,7 +1521,7 @@ async function executeIpAction(
asn: WebsiteIpBlacklist.asn,
})
.from(WebsiteIpBlacklist)
.where(eq(WebsiteIpBlacklist.id, BigInt(id)))
.where(eq(WebsiteIpBlacklist.id, id))
.limit(1)
: await tx
.select({
@@ -1347,21 +1530,17 @@ async function executeIpAction(
asn: WebsiteIpWhitelist.asn,
})
.from(WebsiteIpWhitelist)
.where(eq(WebsiteIpWhitelist.id, BigInt(id)))
.where(eq(WebsiteIpWhitelist.id, id))
.limit(1);
if (blacklist) {
await tx
.delete(WebsiteIpBlacklist)
.where(eq(WebsiteIpBlacklist.id, BigInt(id)));
await tx.delete(WebsiteIpBlacklist).where(eq(WebsiteIpBlacklist.id, id));
} else {
await tx
.delete(WebsiteIpWhitelist)
.where(eq(WebsiteIpWhitelist.id, BigInt(id)));
await tx.delete(WebsiteIpWhitelist).where(eq(WebsiteIpWhitelist.id, id));
}
const snapshot: PeopleMutationSnapshot = {
before: existing
? {
id: Number(existing.id),
id: serializeBigIntIdentifier(existing.id),
action,
ipAddress: existing.ipAddress,
asn: existing.asn,
@@ -1374,7 +1553,7 @@ async function executeIpAction(
context,
"ip.action",
"IpRule",
id,
auditTargetId(id),
snapshot,
"success",
),
@@ -1472,20 +1651,20 @@ async function executeWordFilterUpdate(
const data = record(input);
if (data.action === "add") {
const word = normalizedText(data.word, 255);
let id = 0;
let id = 0n;
let snapshot!: PeopleMutationSnapshot;
await db.transaction(async (tx) => {
const [result] = (await tx
.insert(WebsiteWordfilter)
.values({ word })) as unknown as [ResultSetHeader];
id = Number(result.insertId);
snapshot = { before: null, after: { id, word } };
id = positiveBigIntIdentifier(result.insertId);
snapshot = { before: null, after: { id: id.toString(), word } };
await logAudit(
canonicalAuditEntry(
context,
"word-filter.update",
"WordFilter",
id,
auditTargetId(id),
snapshot,
"intent",
),
@@ -1496,11 +1675,11 @@ async function executeWordFilterUpdate(
context,
"word-filter.update",
"WordFilter",
id,
auditTargetId(id),
snapshot,
async () => {
reloadWordFilter();
await rcon.updateWordFilter();
await requireRcon(await rcon.updateWordFilter());
},
);
}
@@ -1510,20 +1689,18 @@ async function executeWordFilterUpdate(
"errors.housekeeping.validation",
);
}
const id = positiveInteger(data.id);
const id = positiveBigIntIdentifier(data.id);
let snapshot!: PeopleMutationSnapshot;
await db.transaction(async (tx) => {
const [existing] = await tx
.select({ id: WebsiteWordfilter.id, word: WebsiteWordfilter.word })
.from(WebsiteWordfilter)
.where(eq(WebsiteWordfilter.id, BigInt(id)))
.where(eq(WebsiteWordfilter.id, id))
.limit(1);
await tx
.delete(WebsiteWordfilter)
.where(eq(WebsiteWordfilter.id, BigInt(id)));
await tx.delete(WebsiteWordfilter).where(eq(WebsiteWordfilter.id, id));
snapshot = {
before: existing
? { id: Number(existing.id), word: existing.word }
? { id: serializeBigIntIdentifier(existing.id), word: existing.word }
: null,
after: null,
};
@@ -1532,7 +1709,7 @@ async function executeWordFilterUpdate(
context,
"word-filter.update",
"WordFilter",
id,
auditTargetId(id),
snapshot,
"intent",
),
@@ -1543,11 +1720,11 @@ async function executeWordFilterUpdate(
context,
"word-filter.update",
"WordFilter",
id,
auditTargetId(id),
snapshot,
async () => {
reloadWordFilter();
await rcon.updateWordFilter();
await requireRcon(await rcon.updateWordFilter());
},
);
}