jb/eden
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add dataset consent and primary character model

Browse Source
This commit is contained in:
jb
2026-09-19 12:12:10 -04:00
parent a3dc8637f0
commit 4d3c7a7521
23 changed files with 1068 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/convex/functions.ts
+366 -43
View File
@@ -6,6 +6,12 @@ import {
import { v } from "convex/values";
import { assetValidator } from "./schema";
import { capabilityEnabled, parseBackendCapabilityAllowlist } from "../lib/server/release-gate";
import {
CAPABILITY_DEFINITIONS,
SERVER_SCOPE_ALLOWLIST,
validateCapabilityScopes,
} from "../lib/domain/capabilities";
import type { Capability } from "../lib/domain/model";
const MAX_ASSET_CHUNK = 250;
const SYNC_LEASE_MS = 5 * 60 * 1000;
@@ -14,6 +20,17 @@ const MAX_CHUNKS = 4_096;
const MAX_ERROR_LENGTH = 160;
const OAUTH_MAX_LIFETIME_MS = 10 * 60 * 1000;
const SNAPSHOT_MAX_LIFETIME_MS = 60 * 60 * 1000;
const DATASET_SCOPE_REQUIREMENTS: Record<string, readonly string[]> = {
assets: ['esi-assets.read_assets.v1'],
wallet_balance: ['esi-wallet.read_character_wallet.v1'],
wallet_journal: ['esi-wallet.read_character_wallet.v1'],
wallet_transactions: ['esi-wallet.read_character_wallet.v1'],
blueprints: ['esi-characters.read_blueprints.v1'],
skills: ['esi-skills.read_skills.v1', 'esi-skills.read_skillqueue.v1'],
};
const DATASET_NAMES = Object.keys(DATASET_SCOPE_REQUIREMENTS);
const CONSENT_INTENTS = ['initial_login', 'link_character', 'upgrade_dataset', 'reconnect'] as const;
const CONSENT_CAPABILITIES = Object.keys(CAPABILITY_DEFINITIONS) as Capability[];
const bridgeArgs = { bridgeSecret: v.string() };
const characterArgs = { userId: v.string(), characterId: v.number() };
@@ -84,6 +101,106 @@ async function findSessionByHash(ctx: any, sessionHash: string): Promise<any | n
.first();
}
function scopesEqual(left: readonly string[], right: readonly string[]): boolean {
return left.length === right.length && [...left].sort().every((scope, index) => scope === [...right].sort()[index]);
}
function validActualScopes(scopes: readonly string[]): boolean {
return new Set(scopes).size === scopes.length && scopes.every((scope) => SERVER_SCOPE_ALLOWLIST.includes(scope));
}
function datasetStatusesFor(character: any): any[] {
return character.datasetStatuses;
}
function updateDatasetStatuses(
character: any,
scopes: readonly string[],
grantRevision: number,
authorizationEpoch: number,
credentialVersion: number,
timestamp: number,
): { statuses: any[]; assetsInvalidated: boolean } {
const previous = datasetStatusesFor(character);
const statuses: any[] = [];
let assetsInvalidated = false;
for (const dataset of DATASET_NAMES) {
const old = previous.find((entry: any) => entry.dataset === dataset);
const required = DATASET_SCOPE_REQUIREMENTS[dataset]!;
const authorized = required.every((scope) => scopes.includes(scope));
if (!authorized) {
statuses.push({
dataset,
status: 'blocked',
grantRevision,
authorizationEpoch,
credentialVersion,
invalidatedAt: timestamp,
...(dataset === 'assets' ? {} : {}),
});
if (dataset === 'assets' && (old?.status === 'ready' || old?.generation !== undefined || character.activeGeneration !== undefined)) assetsInvalidated = true;
continue;
}
// A newly authorized dataset has no coverage. Existing coverage remains
// usable only when its authority epoch/revision still matches.
const wasCurrent = old && old.grantRevision === grantRevision && old.authorizationEpoch === authorizationEpoch;
const status = wasCurrent ? old.status : old?.status === 'ready' ? 'ready' : 'never';
statuses.push({
dataset,
status,
grantRevision,
authorizationEpoch,
credentialVersion,
...(status === 'ready' && old?.generation === undefined && dataset === 'assets' && character.activeGeneration !== undefined
? { generation: character.activeGeneration } : {}),
...(old?.generation !== undefined && status === 'ready' ? { generation: old.generation } : {}),
...(old?.syncedAt === undefined ? {} : { syncedAt: old.syncedAt }),
...(old?.error === undefined ? {} : { error: old.error }),
});
}
return { statuses, assetsInvalidated };
}
function quarantinedDatasetStatuses(character: any, credentialVersion: number): any[] {
return DATASET_NAMES.map((dataset) => ({
dataset,
status: 'quarantined',
grantRevision: character.grantRevision,
authorizationEpoch: character.authorizationEpoch + 1,
credentialVersion,
invalidatedAt: now(),
}));
}
async function quarantineCharacter(ctx: any, character: any, reason = 'Authorization quarantined'): Promise<void> {
const timestamp = now();
const epoch = character.authorizationEpoch + 1;
await ctx.db.patch(character._id, {
authorizationStatus: 'quarantined',
authorizationEpoch: epoch,
datasetStatuses: quarantinedDatasetStatuses(character, character.credentialVersion),
state: 'error',
activeGeneration: undefined,
syncGeneration: undefined,
syncLeaseExpiresAt: undefined,
error: reason.slice(0, MAX_ERROR_LENGTH),
});
const generations = await ctx.db.query('assetGenerations').withIndex('by_user_character_generation', (q: any) =>
q.eq('userId', character.userId).eq('characterId', character.characterId)).collect();
for (const generation of generations) {
if (generation.status === 'staging') await ctx.db.patch(generation._id, { status: 'failed', error: 'Authorization quarantined' });
}
const jobs = await ctx.db.query('jobs').withIndex('by_user_status', (q: any) => q.eq('userId', character.userId)).collect();
for (const job of jobs) {
if (job.characterId === character.characterId && !['succeeded', 'failed', 'cancelled'].includes(job.status)) {
await ctx.db.patch(job._id, { status: 'quarantined', updatedAt: timestamp, leaseExpiresAt: undefined });
}
}
const token = await ctx.db.query('tokenRecords').withIndex('by_user_character', (q: any) =>
q.eq('userId', character.userId).eq('characterId', character.characterId)).first();
if (token) await ctx.db.delete(token._id);
}
function isUsableSession(session: any, timestamp = now()): boolean {
return Boolean(
session &&
@@ -125,6 +242,12 @@ export const bridgeCreateOAuth = mutationGeneric({
encryptedVerifier: v.string(),
expiresAt: v.number(),
returnTo: v.string(),
intent: v.string(),
capability: v.string(),
requestedScopes: v.array(v.string()),
approvedExistingScopes: v.array(v.string()),
initiatingSessionHash: v.optional(v.string()),
targetCharacterId: v.optional(v.number()),
},
handler: async (ctx, args) => {
assertBridge(args.bridgeSecret);
@@ -136,6 +259,12 @@ export const bridgeCreateOAuth = mutationGeneric({
) {
throw new Error("Invalid OAuth state expiry");
}
if (!CONSENT_INTENTS.includes(args.intent as typeof CONSENT_INTENTS[number])) throw new Error("Invalid OAuth intent");
if (!CONSENT_CAPABILITIES.includes(args.capability as Capability)) throw new Error("Invalid OAuth capability");
if (!validActualScopes(args.requestedScopes)) throw new Error("Invalid OAuth requested scopes");
if (!validActualScopes(args.approvedExistingScopes)) throw new Error("Invalid OAuth preserved scopes");
const scopeCheck = validateCapabilityScopes(args.capability as Capability, args.requestedScopes, args.approvedExistingScopes);
if (!scopeCheck.valid) throw new Error("Invalid OAuth capability scopes");
// Keep this bounded cleanup opportunistic; retention/GC is intentionally
// outside this bridge transaction.
const expired = await ctx.db
@@ -156,6 +285,12 @@ export const bridgeCreateOAuth = mutationGeneric({
encryptedVerifier: args.encryptedVerifier,
expiresAt: args.expiresAt,
returnTo: args.returnTo,
intent: args.intent,
capability: args.capability,
requestedScopes: args.requestedScopes,
approvedExistingScopes: args.approvedExistingScopes,
...(args.initiatingSessionHash === undefined ? {} : { initiatingSessionHash: args.initiatingSessionHash }),
...(args.targetCharacterId === undefined ? {} : { targetCharacterId: args.targetCharacterId }),
});
return null;
},
@@ -183,6 +318,12 @@ export const bridgeConsumeOAuth = mutationGeneric({
encryptedVerifier: state.encryptedVerifier,
expiresAt: state.expiresAt,
returnTo: state.returnTo,
intent: state.intent,
capability: state.capability,
requestedScopes: state.requestedScopes,
approvedExistingScopes: state.approvedExistingScopes,
...(state.initiatingSessionHash === undefined ? {} : { initiatingSessionHash: state.initiatingSessionHash }),
...(state.targetCharacterId === undefined ? {} : { targetCharacterId: state.targetCharacterId }),
};
},
});
@@ -198,6 +339,12 @@ const completeLoginArgs = {
tokenExpiresAt: v.number(),
sessionHash: v.string(),
sessionExpiresAt: v.number(),
intent: v.string(),
capability: v.string(),
requestedScopes: v.array(v.string()),
approvedExistingScopes: v.array(v.string()),
initiatingSessionHash: v.optional(v.string()),
targetCharacterId: v.optional(v.number()),
};
/**
@@ -209,83 +356,131 @@ export const bridgeCompleteLogin = mutationGeneric({
args: completeLoginArgs,
handler: async (ctx, args) => {
assertBridge(args.bridgeSecret);
const timestamp = now();
if (!validActualScopes(args.scopes)) throw new Error("Invalid granted scopes");
if (args.scopes.some((scope) => !args.requestedScopes.includes(scope))) throw new Error("Granted scopes exceed the bound request");
if (!CONSENT_INTENTS.includes(args.intent as typeof CONSENT_INTENTS[number])) throw new Error("Invalid OAuth intent");
if (!CONSENT_CAPABILITIES.includes(args.capability as Capability)) throw new Error("Invalid OAuth capability");
if (!validActualScopes(args.requestedScopes)) throw new Error("Invalid OAuth requested scopes");
if (!validActualScopes(args.approvedExistingScopes)) throw new Error("Invalid OAuth preserved scopes");
const scopeCheck = validateCapabilityScopes(args.capability as Capability, args.requestedScopes, args.approvedExistingScopes);
if (!scopeCheck.valid) throw new Error("Invalid OAuth capability scopes");
if (args.targetCharacterId !== undefined && args.targetCharacterId !== args.characterId) {
throw new Error("OAuth character mismatch");
}
const byCharacterId = await ctx.db
.query("characters")
.withIndex("by_character_id", (q: any) => q.eq("characterId", args.characterId))
.collect();
if (byCharacterId.some((character: any) => character.ownerHash !== args.ownerHash)) {
throw new Error("Character ownership conflict");
const conflicting = byCharacterId.filter((character: any) => character.ownerHash !== args.ownerHash);
if (conflicting.length > 0) {
// Never attach a provider character to a new owner. Quarantine every
// old row in this transaction; recovery/transfer is not part of this
// local prototype.
for (const oldCharacter of conflicting) {
await quarantineCharacter(ctx, oldCharacter, "Character ownership changed; reconnect required");
}
// Return a committed rejection so quarantine is not rolled back by a
// thrown Convex error. The server adapter converts this to a fixed
// callback failure without issuing a session.
return { userId: conflicting[0]!.userId, sessionId: "", rejected: "ownership_conflict" };
}
const existingCharacter = byCharacterId[0] ?? null;
if (args.intent === "link_character" && existingCharacter) {
throw new Error("Character is already linked; use reconnect or upgrade");
}
let resolvedUserId: string;
if (existingCharacter) {
if (existingCharacter.ownerHash !== args.ownerHash) {
throw new Error("Character ownership conflict");
}
if (args.userId !== undefined && args.userId !== existingCharacter.userId) {
throw new Error("Character belongs to another user");
}
resolvedUserId = existingCharacter.userId;
if (args.userId === undefined && existingCharacter.loginRole === "data_only") {
throw new Error("Data-only character cannot authenticate while signed out");
}
} else {
// A signed-out first login creates one app user for this character. A
// new character may only be attached when the current app user is
// explicitly supplied by the authenticated linking flow.
resolvedUserId = args.userId ?? newIdentifier("user");
}
const sameSession = await findSessionByHash(ctx, args.sessionHash);
if (sameSession) {
throw new Error("Session hash collision");
if (sameSession) throw new Error("Session hash collision");
// Linking, upgrading and reconnecting are authorized by the still-live
// application session that initiated consent.
const preexistingUser = await ctx.db.query("users").withIndex("by_user_id", (q: any) => q.eq("userId", resolvedUserId)).first();
if (args.userId !== undefined && !preexistingUser) throw new Error("User does not exist");
let initiatingSession: any | null = null;
if (args.userId !== undefined) {
initiatingSession = args.initiatingSessionHash
? await findSessionByHash(ctx, args.initiatingSessionHash)
: null;
if (!isUsableSession(initiatingSession, timestamp) || initiatingSession.userId !== args.userId) {
throw new Error("Initiating session is no longer valid");
}
}
if (args.intent === "upgrade_dataset" || args.intent === "reconnect") {
if (args.userId === undefined || args.targetCharacterId !== args.characterId) throw new Error("OAuth target is required");
}
if (args.intent === "link_character" && args.userId === undefined) throw new Error("Authenticated session is required to link a character");
const user = await ctx.db
.query("users")
.withIndex("by_user_id", (q: any) => q.eq("userId", resolvedUserId))
.first();
if (!user) {
if (args.userId !== undefined || existingCharacter) {
throw new Error("User does not exist");
}
await ctx.db.insert("users", { userId: resolvedUserId, createdAt: now() });
if (args.userId !== undefined || existingCharacter) throw new Error("User does not exist");
await ctx.db.insert("users", { userId: resolvedUserId, createdAt: timestamp });
}
let character = existingCharacter;
if (!character) {
character = await findCharacter(ctx, resolvedUserId, args.characterId);
}
if (!character) character = await findCharacter(ctx, resolvedUserId, args.characterId);
const oldScopes = character?.scopes ?? [];
const oldRevision = character?.grantRevision ?? 1;
const grantChanged = character ? !scopesEqual(oldScopes, args.scopes) : false;
const grantRevision = character ? oldRevision + (grantChanged ? 1 : 0) : 1;
const authorizationEpoch = character?.authorizationEpoch ?? 1;
const existingTokenForVersion = await ctx.db.query("tokenRecords").withIndex("by_user_character", (q: any) => q.eq("userId", resolvedUserId).eq("characterId", args.characterId)).first();
const credentialVersion = Math.max(character?.credentialVersion ?? 0, existingTokenForVersion?.version ?? 0) + 1;
const updated = character
? updateDatasetStatuses(character, args.scopes, grantRevision, authorizationEpoch, credentialVersion, timestamp)
: updateDatasetStatuses({ state: "never", scopes: [], datasetStatuses: [] }, args.scopes, grantRevision, authorizationEpoch, credentialVersion, timestamp);
const basePatch = {
name: args.name,
scopes: args.scopes,
ownerHash: args.ownerHash,
loginRole: character ? character.loginRole : (args.userId === undefined ? "primary" : "data_only"),
authorizationStatus: "active" as const,
grantRevision,
credentialVersion,
authorizationEpoch,
datasetStatuses: updated.statuses,
...(updated.assetsInvalidated ? { state: "error" as const, activeGeneration: undefined, syncedAt: undefined, expiresAt: undefined } : {}),
};
if (character) {
if (character.ownerHash !== args.ownerHash || character.userId !== resolvedUserId) {
throw new Error("Character ownership conflict");
}
await ctx.db.patch(character._id, {
name: args.name,
scopes: args.scopes,
ownerHash: args.ownerHash,
});
await ctx.db.patch(character._id, basePatch);
} else {
await ctx.db.insert("characters", {
userId: resolvedUserId,
characterId: args.characterId,
name: args.name,
ownerHash: args.ownerHash,
scopes: args.scopes,
state: "never",
...basePatch,
});
}
const tokenRecord = await ctx.db
.query("tokenRecords")
.withIndex("by_user_character", (q: any) =>
q.eq("userId", resolvedUserId).eq("characterId", args.characterId),
)
.withIndex("by_user_character", (q: any) => q.eq("userId", resolvedUserId).eq("characterId", args.characterId))
.first();
if (tokenRecord) {
await ctx.db.patch(tokenRecord._id, {
encryptedTokens: args.encryptedTokens,
tokenExpiresAt: args.tokenExpiresAt,
scopes: args.scopes,
version: tokenRecord.version + 1,
version: credentialVersion,
});
} else {
await ctx.db.insert("tokenRecords", {
@@ -294,7 +489,7 @@ export const bridgeCompleteLogin = mutationGeneric({
encryptedTokens: args.encryptedTokens,
tokenExpiresAt: args.tokenExpiresAt,
scopes: args.scopes,
version: 1,
version: credentialVersion,
});
}
@@ -305,6 +500,9 @@ export const bridgeCompleteLogin = mutationGeneric({
userId: resolvedUserId,
expiresAt: args.sessionExpiresAt,
});
// Replacement is atomic with persistence; the callback never has to
// replay a code if cookie delivery later fails.
if (initiatingSession && !initiatingSession.revokedAt) await ctx.db.patch(initiatingSession._id, { revokedAt: timestamp });
return { userId: resolvedUserId, sessionId };
},
});
@@ -314,7 +512,7 @@ export const bridgeTokenRecord = queryGeneric({
handler: async (ctx, args) => {
assertBridge(args.bridgeSecret);
const character = await findCharacter(ctx, args.userId, args.characterId);
if (!character) return null;
if (!character || character.authorizationStatus !== "active") return null;
return await ctx.db
.query("tokenRecords")
.withIndex("by_user_character", (q: any) =>
@@ -345,8 +543,9 @@ export const bridgeSaveTokens = mutationGeneric({
},
handler: async (ctx, args) => {
assertBridge(args.bridgeSecret);
if (!validActualScopes(args.scopes)) return false;
const character = await findCharacter(ctx, args.userId, args.characterId);
if (!character) return false;
if (!character || character.authorizationStatus !== "active") return false;
const record = await ctx.db
.query("tokenRecords")
.withIndex("by_user_character", (q: any) =>
@@ -354,13 +553,22 @@ export const bridgeSaveTokens = mutationGeneric({
)
.first();
if (!record || record.version !== args.expectedVersion) return false;
const nextVersion = record.version + 1;
const nextRevision = character.grantRevision + (scopesEqual(character.scopes, args.scopes) ? 0 : 1);
const updated = updateDatasetStatuses(character, args.scopes, nextRevision, character.authorizationEpoch, nextVersion, now());
await ctx.db.patch(record._id, {
encryptedTokens: args.encryptedTokens,
tokenExpiresAt: args.tokenExpiresAt,
scopes: args.scopes,
version: record.version + 1,
version: nextVersion,
});
await ctx.db.patch(character._id, {
scopes: args.scopes,
grantRevision: nextRevision,
credentialVersion: nextVersion,
datasetStatuses: updated.statuses,
...(updated.assetsInvalidated ? { state: "error" as const, activeGeneration: undefined, syncedAt: undefined, expiresAt: undefined } : {}),
});
await ctx.db.patch(character._id, { scopes: args.scopes });
return true;
},
});
@@ -376,6 +584,84 @@ export const bridgeCharacters = queryGeneric({
},
});
/** Change the one signed-out login character without changing authorization. */
export const bridgeSetPrimary = mutationGeneric({
args: { ...bridgeArgs, userId: v.string(), characterId: v.number(), initiatingSessionHash: v.string() },
handler: async (ctx, args) => {
assertBridge(args.bridgeSecret);
const initiating = await findSessionByHash(ctx, args.initiatingSessionHash);
if (!isUsableSession(initiating) || initiating.userId !== args.userId) throw new Error("Session is no longer valid");
const target = await findCharacter(ctx, args.userId, args.characterId);
const targetToken = target
? await ctx.db.query("tokenRecords").withIndex("by_user_character", (q: any) => q.eq("userId", args.userId).eq("characterId", args.characterId)).first()
: null;
if (!target || !targetToken || target.authorizationStatus !== "active") {
throw new Error("Character is not eligible to be primary");
}
const characters = await ctx.db.query("characters").withIndex("by_user_id", (q: any) => q.eq("userId", args.userId)).collect();
for (const character of characters) {
if (character._id === target._id) {
await ctx.db.patch(character._id, { loginRole: "primary" });
} else if (character.loginRole === "primary" || character.loginRole === undefined) {
await ctx.db.patch(character._id, { loginRole: "data_only" });
}
}
return null;
},
});
/** Local cancellation immediately fences private data and future work. */
export const bridgeQuarantineAuthorization = mutationGeneric({
args: { ...bridgeArgs, userId: v.string(), characterId: v.number() },
handler: async (ctx, args) => {
assertBridge(args.bridgeSecret);
const character = await findCharacter(ctx, args.userId, args.characterId);
if (character) await quarantineCharacter(ctx, character, "Character ownership changed; reconnect required");
return null;
},
});
export const bridgeInvalidateAuthorization = mutationGeneric({
args: { ...bridgeArgs, userId: v.string(), characterId: v.number(), initiatingSessionHash: v.string() },
handler: async (ctx, args) => {
assertBridge(args.bridgeSecret);
const initiating = await findSessionByHash(ctx, args.initiatingSessionHash);
if (!isUsableSession(initiating) || initiating.userId !== args.userId) throw new Error("Session is no longer valid");
const character = await findCharacter(ctx, args.userId, args.characterId);
if (!character) throw new Error("Character not found");
if (character.loginRole === "primary") throw new Error("Switch the primary character before cancelling its authorization");
const timestamp = now();
const epoch = character.authorizationEpoch + 1;
const statuses = DATASET_NAMES.map((dataset) => ({
dataset,
status: "blocked" as const,
grantRevision: character.grantRevision,
authorizationEpoch: epoch,
credentialVersion: character.credentialVersion,
invalidatedAt: timestamp,
}));
await ctx.db.patch(character._id, {
authorizationStatus: "revoked",
authorizationEpoch: epoch,
datasetStatuses: statuses,
state: "error",
activeGeneration: undefined,
syncGeneration: undefined,
syncLeaseExpiresAt: undefined,
error: "Authorization cancelled",
});
const token = await ctx.db.query("tokenRecords").withIndex("by_user_character", (q: any) => q.eq("userId", args.userId).eq("characterId", args.characterId)).first();
if (token) await ctx.db.delete(token._id);
const jobs = await ctx.db.query("jobs").withIndex("by_user_status", (q: any) => q.eq("userId", args.userId)).collect();
for (const job of jobs) {
if (job.characterId === args.characterId && !["succeeded", "failed", "cancelled"].includes(job.status)) {
await ctx.db.patch(job._id, { status: "quarantined", updatedAt: timestamp, leaseExpiresAt: undefined });
}
}
return null;
},
});
/**
* Read an active snapshot one page of chunks at a time. The application adapter
* walks this query's cursor, rather than issuing an unbounded Convex collect.
@@ -391,7 +677,14 @@ export const bridgeAssetsPage = queryGeneric({
handler: async (ctx, args) => {
assertBridge(args.bridgeSecret);
const character = await findCharacter(ctx, args.userId, args.characterId);
const generation = character?.activeGeneration ?? null;
const assetStatus = character?.datasetStatuses?.find((entry: any) => entry.dataset === "assets");
const authorized = character && character.authorizationStatus === "active"
&& character.scopes.includes("esi-assets.read_assets.v1")
&& assetStatus
&& (assetStatus.status === "ready" || assetStatus.status === "error")
&& assetStatus.grantRevision === character.grantRevision
&& assetStatus.authorizationEpoch === character.authorizationEpoch;
const generation = authorized ? character.activeGeneration ?? null : null;
if (!character || !generation) {
return { page: [], isDone: true, continueCursor: "", generation: null };
}
@@ -426,6 +719,9 @@ export const bridgeBeginSync = mutationGeneric({
assertDeploymentCapability('asset_sync');
const character = await findCharacter(ctx, args.userId, args.characterId);
if (!character) throw new Error("Character not found");
if (character.authorizationStatus !== "active" || !character.scopes.includes("esi-assets.read_assets.v1")) {
throw new Error("Asset authorization is not active");
}
const timestamp = now();
if (character.syncGeneration && (character.syncLeaseExpiresAt ?? 0) > timestamp) {
throw new Error("Character synchronization is already leased");
@@ -437,9 +733,16 @@ export const bridgeBeginSync = mutationGeneric({
generation,
status: "staging",
stagedChunkCount: 0,
grantRevision: character.grantRevision,
authorizationEpoch: character.authorizationEpoch,
credentialVersion: character.credentialVersion,
});
const statuses = datasetStatusesFor(character).map((entry: any) => entry.dataset === "assets"
? { ...entry, status: "syncing", grantRevision: character.grantRevision, authorizationEpoch: character.authorizationEpoch }
: entry);
await ctx.db.patch(character._id, {
state: "syncing",
datasetStatuses: statuses,
syncGeneration: generation,
syncLeaseExpiresAt: timestamp + SYNC_LEASE_MS,
error: undefined,
@@ -565,14 +868,24 @@ export const bridgeFinishSync = mutationGeneric({
throw new Error("Snapshot is missing chunks");
}
if (generation.grantRevision !== character.grantRevision
|| generation.authorizationEpoch !== character.authorizationEpoch
|| generation.credentialVersion !== character.credentialVersion
|| character.authorizationStatus !== "active") {
throw new Error("Synchronization authorization is stale");
}
await ctx.db.patch(generation._id, {
status: "complete",
chunkCount: args.chunkCount,
syncedAt: args.syncedAt,
expiresAt: args.expiresAt,
});
const statuses = datasetStatusesFor(character).map((entry: any) => entry.dataset === "assets"
? { ...entry, status: "ready", generation: args.generation, syncedAt: args.syncedAt, error: undefined }
: entry);
await ctx.db.patch(character._id, {
state: "ready",
datasetStatuses: statuses,
syncedAt: args.syncedAt,
expiresAt: args.expiresAt,
activeGeneration: args.generation,
@@ -602,16 +915,26 @@ export const bridgeFailSync = mutationGeneric({
.eq("generation", args.generation),
)
.first();
if (generation) {
await ctx.db.patch(generation._id, {
status: "failed",
error: safeSyncError(args.error).slice(0, MAX_ERROR_LENGTH),
});
if (!generation || generation.status !== "staging"
|| (character.syncLeaseExpiresAt ?? 0) <= now()
|| generation.grantRevision !== character.grantRevision
|| generation.authorizationEpoch !== character.authorizationEpoch
|| generation.credentialVersion !== character.credentialVersion
|| character.authorizationStatus !== "active") {
throw new Error("Synchronization failure publication is stale");
}
await ctx.db.patch(generation._id, {
status: "failed",
error: safeSyncError(args.error).slice(0, MAX_ERROR_LENGTH),
});
// Deliberately do not touch activeGeneration: the previous complete
// snapshot remains available after a failed synchronization.
const statuses = datasetStatusesFor(character).map((entry: any) => entry.dataset === "assets"
? { ...entry, status: character.activeGeneration ? "error" : "blocked", error: safeSyncError(args.error).slice(0, MAX_ERROR_LENGTH) }
: entry);
await ctx.db.patch(character._id, {
state: "error",
datasetStatuses: statuses,
error: safeSyncError(args.error).slice(0, MAX_ERROR_LENGTH),
syncGeneration: undefined,
syncLeaseExpiresAt: undefined,